diff --git a/config/app.go b/config/app.go index 67aafc4..7700e31 100644 --- a/config/app.go +++ b/config/app.go @@ -19,6 +19,7 @@ type AppConfig struct { HostKeyPath string `yaml:"host_key_path"` DBPath string `yaml:"db_path"` Origin string `yaml:"origin"` + LogLevel string `yaml:"log_level"` SMTP SMTPConfig `yaml:"smtp"` AllowAllKeys bool `yaml:"allow_all_keys"` AllowedKeys []string `yaml:"allowed_keys"` @@ -44,6 +45,7 @@ func DefaultAppConfig() *AppConfig { HostKeyPath: "./host_key", DBPath: "./herald.db", Origin: "http://localhost:8080", + LogLevel: "info", SMTP: SMTPConfig{ Host: "localhost", Port: 587, @@ -165,4 +167,7 @@ func applyEnvOverrides(cfg *AppConfig) { if v := os.Getenv("HERALD_ORIGIN"); v != "" { cfg.Origin = v } + if v := os.Getenv("HERALD_LOG_LEVEL"); v != "" { + cfg.LogLevel = v + } } diff --git a/email/send.go b/email/send.go index 2e6f611..7d8f64c 100644 --- a/email/send.go +++ b/email/send.go @@ -257,7 +257,7 @@ func (m *Mailer) Send(to, subject, htmlBody, textBody, unsubToken, dashboardURL, return m.sendWithTLS(addr, auth, to, messageBytes) } - return smtp.SendMail(addr, auth, m.cfg.From, []string{to}, messageBytes) + return m.sendWithSTARTTLS(addr, auth, to, messageBytes) } func encodeQuotedPrintable(s string) string { @@ -274,10 +274,63 @@ func (m *Mailer) sendWithTLS(addr string, auth smtp.Auth, to string, msg []byte) MinVersion: tls.VersionTLS12, } - conn, err := tls.Dial("tcp", addr, tlsConfig) + dialer := &net.Dialer{Timeout: 30 * time.Second} + conn, err := tls.DialWithDialer(dialer, "tcp", addr, tlsConfig) if err != nil { return fmt.Errorf("TLS dial: %w", err) } + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + _ = conn.Close() + return fmt.Errorf("set deadline: %w", err) + } + defer func() { _ = conn.Close() }() + + client, err := smtp.NewClient(conn, m.cfg.Host) + if err != nil { + return fmt.Errorf("SMTP client: %w", err) + } + defer func() { _ = client.Close() }() + + if auth != nil { + if err = client.Auth(auth); err != nil { + return fmt.Errorf("auth: %w", err) + } + } + + if err = client.Mail(m.cfg.From); err != nil { + return fmt.Errorf("mail from: %w", err) + } + + if err = client.Rcpt(to); err != nil { + return fmt.Errorf("rcpt to: %w", err) + } + + w, err := client.Data() + if err != nil { + return fmt.Errorf("data: %w", err) + } + + if _, err = w.Write(msg); err != nil { + return fmt.Errorf("write: %w", err) + } + + if err = w.Close(); err != nil { + return fmt.Errorf("close data: %w", err) + } + + return client.Quit() +} + +func (m *Mailer) sendWithSTARTTLS(addr string, auth smtp.Auth, to string, msg []byte) error { + dialer := &net.Dialer{Timeout: 30 * time.Second} + conn, err := dialer.Dial("tcp", addr) + if err != nil { + return fmt.Errorf("dial: %w", err) + } + if err := conn.SetDeadline(time.Now().Add(30 * time.Second)); err != nil { + _ = conn.Close() + return fmt.Errorf("set deadline: %w", err) + } defer func() { _ = conn.Close() }() client, err := smtp.NewClient(conn, m.cfg.Host) @@ -286,6 +339,13 @@ func (m *Mailer) sendWithTLS(addr string, auth smtp.Auth, to string, msg []byte) } defer func() { _ = client.Close() }() + if err = client.StartTLS(&tls.Config{ + ServerName: m.cfg.Host, + MinVersion: tls.VersionTLS12, + }); err != nil { + return fmt.Errorf("STARTTLS: %w", err) + } + if auth != nil { if err = client.Auth(auth); err != nil { return fmt.Errorf("auth: %w", err) diff --git a/main.go b/main.go index 9c4ae8c..593838f 100644 --- a/main.go +++ b/main.go @@ -130,6 +130,20 @@ func runServer(ctx context.Context) error { return fmt.Errorf("failed to load config: %w", err) } + // Set log level from config + level := log.InfoLevel + switch strings.ToLower(cfg.LogLevel) { + case "debug": + level = log.DebugLevel + case "info": + level = log.InfoLevel + case "warn": + level = log.WarnLevel + case "error": + level = log.ErrorLevel + } + logger.SetLevel(level) + logger.Info("starting herald", "ssh_port", cfg.SSHPort, "http_port", cfg.HTTPPort, diff --git a/scheduler/fetch.go b/scheduler/fetch.go index 8989f12..fd8cfc4 100644 --- a/scheduler/fetch.go +++ b/scheduler/fetch.go @@ -5,6 +5,7 @@ import ( "context" "net/http" "sync" + "sync/atomic" "time" "github.com/kierank/herald/store" @@ -12,8 +13,8 @@ import ( ) const ( - feedFetchTimeout = 30 * time.Second - maxConcurrentFetch = 10 + feedFetchTimeout = 15 * time.Second + maxConcurrentFetch = 30 ) type FetchResult struct { @@ -63,7 +64,7 @@ func FetchFeed(ctx context.Context, feed *store.Feed) *FetchResult { } client := &http.Client{ - Timeout: 30 * time.Second, + Timeout: 15 * time.Second, } resp, err := client.Do(req) @@ -125,7 +126,7 @@ func FetchFeed(ctx context.Context, feed *store.Feed) *FetchResult { return result } -func FetchFeeds(ctx context.Context, feeds []*store.Feed) []*FetchResult { +func FetchFeeds(ctx context.Context, feeds []*store.Feed, progress *atomic.Int32) []*FetchResult { results := make([]*FetchResult, len(feeds)) var wg sync.WaitGroup @@ -138,7 +139,12 @@ func FetchFeeds(ctx context.Context, feeds []*store.Feed) []*FetchResult { for i, feed := range feeds { wg.Add(1) go func(idx int, f *store.Feed) { - defer wg.Done() + defer func() { + if progress != nil { + progress.Add(1) + } + wg.Done() + }() semaphore <- struct{}{} // Acquire defer func() { <-semaphore }() // Release results[idx] = FetchFeed(ctx, f) diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index 295c8c5..267c0c4 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -3,6 +3,7 @@ package scheduler import ( "context" "fmt" + "sync/atomic" "time" "github.com/adhocore/gronx" @@ -32,6 +33,15 @@ const ( minSendsBeforeDeactivate = 3 // minimum sends before considering deactivation ) +// RunStats contains detailed statistics from a feed fetch run +type RunStats struct { + TotalFeeds int + FetchedFeeds int + FailedFeeds int + NewItems int + EmailSent bool +} + type Scheduler struct { store *store.DB mailer *email.Mailer @@ -184,36 +194,59 @@ func (s *Scheduler) tick(ctx context.Context) { } } -func (s *Scheduler) RunNow(ctx context.Context, configID int64) (int, error) { +func (s *Scheduler) RunNow(ctx context.Context, configID int64, progress *atomic.Int32) (*RunStats, error) { cfg, err := s.store.GetConfigByID(ctx, configID) if err != nil { - return 0, fmt.Errorf("get config: %w", err) + return nil, fmt.Errorf("get config: %w", err) } feeds, err := s.store.GetFeedsByConfig(ctx, cfg.ID) if err != nil { - return 0, fmt.Errorf("get feeds: %w", err) + return nil, fmt.Errorf("get feeds: %w", err) } if len(feeds) == 0 { - return 0, fmt.Errorf("no feeds configured") + return nil, fmt.Errorf("no feeds configured") } - results := FetchFeeds(ctx, feeds) + stats := &RunStats{ + TotalFeeds: len(feeds), + } + + results := FetchFeeds(ctx, feeds, progress) + s.logger.Debug("RunNow: fetching complete", "total", len(feeds)) + + // Count successful and failed fetches + for _, result := range results { + if result.Error != nil { + stats.FailedFeeds++ + } else { + stats.FetchedFeeds++ + } + } + s.logger.Debug("RunNow: counting complete", "fetched", stats.FetchedFeeds, "failed", stats.FailedFeeds) feedGroups, totalNew, err := s.collectNewItems(ctx, results) + s.logger.Debug("RunNow: collectNewItems complete", "totalNew", totalNew, "err", err) if err != nil { - return 0, err + return stats, err } + stats.NewItems = totalNew + if totalNew > 0 { + s.logger.Debug("RunNow: starting email send") if err := s.sendDigestAndMarkSeen(ctx, cfg, feedGroups, totalNew, results); err != nil { - return 0, err + s.logger.Error("RunNow: sendDigestAndMarkSeen failed", "err", err) + return stats, err } + stats.EmailSent = true s.logger.Info("email sent", "to", cfg.Email, "items", totalNew) } + s.logger.Debug("RunNow: email phase complete") // Update feed metadata + s.logger.Debug("RunNow: updating feed metadata", "count", len(results)) for _, result := range results { if result.ETag != "" || result.LastModified != "" { if err := s.store.UpdateFeedFetched(ctx, result.FeedID, result.ETag, result.LastModified); err != nil { @@ -221,20 +254,24 @@ func (s *Scheduler) RunNow(ctx context.Context, configID int64) (int, error) { } } } + s.logger.Debug("RunNow: feed metadata updated") + s.logger.Debug("RunNow: calculating next run") now := time.Now() nextRun, err := gronx.NextTick(cfg.CronExpr, false) if err != nil { - return totalNew, fmt.Errorf("calculate next run: %w", err) + return stats, fmt.Errorf("calculate next run: %w", err) } + s.logger.Debug("RunNow: updating last run", "nextRun", nextRun) if err := s.store.UpdateLastRun(ctx, cfg.ID, now, nextRun); err != nil { - return totalNew, fmt.Errorf("update last run: %w", err) + return stats, fmt.Errorf("update last run: %w", err) } _ = s.store.AddLog(ctx, cfg.ID, "info", fmt.Sprintf("Processed: %d new items, next run: %s", totalNew, nextRun.Format(time.RFC3339))) - return totalNew, nil + s.logger.Debug("RunNow: complete") + return stats, nil } func (s *Scheduler) collectNewItems(ctx context.Context, results []*FetchResult) ([]email.FeedGroup, int, error) { @@ -305,6 +342,7 @@ func (s *Scheduler) collectNewItems(ctx context.Context, results []*FetchResult) } func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config, feedGroups []email.FeedGroup, totalNew int, results []*FetchResult) error { + s.logger.Debug("sendDigestAndMarkSeen: start", "totalNew", totalNew) digestData := &email.DigestData{ ConfigName: cfg.Filename, TotalItems: totalNew, @@ -316,16 +354,19 @@ func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config inline = false } + s.logger.Debug("sendDigestAndMarkSeen: rendering digest") htmlBody, textBody, err := email.RenderDigest(digestData, inline) if err != nil { return fmt.Errorf("render digest: %w", err) } + s.logger.Debug("sendDigestAndMarkSeen: digest rendered") unsubToken, err := s.store.GetOrCreateUnsubscribeToken(ctx, cfg.ID) if err != nil { s.logger.Warn("failed to create unsubscribe token", "err", err) unsubToken = "" } + s.logger.Debug("sendDigestAndMarkSeen: got unsub token") user, err := s.store.GetUserByID(ctx, cfg.UserID) dashboardURL := "" @@ -334,11 +375,13 @@ func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config } else { s.logger.Warn("failed to get user for dashboard URL", "err", err) } + s.logger.Debug("sendDigestAndMarkSeen: got dashboard URL") // Rate limit email sending per user if !s.rateLimiter.Allow(fmt.Sprintf("email:%d", cfg.UserID)) { return fmt.Errorf("rate limit exceeded for email sending") } + s.logger.Debug("sendDigestAndMarkSeen: rate limit ok") // Begin transaction to mark items seen tx, err := s.store.BeginTx(ctx) @@ -346,6 +389,7 @@ func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config return fmt.Errorf("begin transaction: %w", err) } defer func() { _ = tx.Rollback() }() + s.logger.Debug("sendDigestAndMarkSeen: transaction started") // Mark items seen BEFORE sending email for _, result := range results { @@ -358,20 +402,31 @@ func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config } } } + s.logger.Debug("sendDigestAndMarkSeen: items marked seen") - // Send email - if this fails, transaction will rollback - subject := "feed digest" - - // Record email send with tracking - trackingToken, err := s.store.RecordEmailSend(cfg.ID, cfg.Email, subject, true) + // Generate tracking token BEFORE recording (needed for email pixel URL) + trackingToken, err := s.store.GenerateTrackingToken() if err != nil { - s.logger.Warn("failed to record email send", "err", err) + s.logger.Warn("failed to generate tracking token", "err", err) trackingToken = "" } + s.logger.Debug("sendDigestAndMarkSeen: generated tracking token") + + // Record email send with tracking (within transaction) + subject := "feed digest" + s.logger.Debug("sendDigestAndMarkSeen: recording email send") + if err := s.store.RecordEmailSendTx(tx, cfg.ID, cfg.Email, subject, trackingToken); err != nil { + s.logger.Warn("failed to record email send", "err", err) + } + s.logger.Debug("sendDigestAndMarkSeen: recorded email send") + // Send email - if this fails, transaction will rollback + s.logger.Debug("sendDigestAndMarkSeen: calling mailer.Send", "to", cfg.Email) if err := s.mailer.Send(cfg.Email, subject, htmlBody, textBody, unsubToken, dashboardURL, trackingToken); err != nil { + s.logger.Error("sendDigestAndMarkSeen: mailer.Send failed", "err", err) return fmt.Errorf("send email: %w", err) } + s.logger.Debug("sendDigestAndMarkSeen: mailer.Send returned successfully") // Commit transaction only after successful email send if err := tx.Commit(); err != nil { @@ -394,7 +449,7 @@ func (s *Scheduler) processConfig(ctx context.Context, cfg *store.Config) error return nil } - results := FetchFeeds(ctx, feeds) + results := FetchFeeds(ctx, feeds, nil) // No progress tracking for background jobs feedGroups, totalNew, err := s.collectNewItems(ctx, results) if err != nil { diff --git a/ssh/commands.go b/ssh/commands.go index 1a36f70..92691c6 100644 --- a/ssh/commands.go +++ b/ssh/commands.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "strings" + "sync/atomic" "time" "github.com/charmbracelet/lipgloss" @@ -175,15 +176,24 @@ func handleRun(ctx context.Context, sess ssh.Session, user *store.User, st *stor return } - // Simple spinner animation + // Get feed count for progress display + feeds, err := st.GetFeedsByConfig(ctx, cfg.ID) + if err != nil { + println(sess, errorStyle.Render("Error: "+err.Error())) + return + } + totalFeeds := len(feeds) + + // Progress tracking + var progress atomic.Int32 spinChars := []string{"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"} done := make(chan struct{}) result := make(chan struct { - items int + stats *scheduler.RunStats err error }) - // Spinner goroutine + // Spinner goroutine with real-time progress go func() { i := 0 for { @@ -191,7 +201,8 @@ func handleRun(ctx context.Context, sess ssh.Session, user *store.User, st *stor case <-done: return default: - printf(sess, "\r%s Fetching feeds...", spinChars[i%len(spinChars)]) + completed := progress.Load() + printf(sess, "\r%s Fetching feeds... %d/%d", spinChars[i%len(spinChars)], completed, totalFeeds) i++ time.Sleep(80 * time.Millisecond) } @@ -200,11 +211,11 @@ func handleRun(ctx context.Context, sess ssh.Session, user *store.User, st *stor // Work goroutine go func() { - newItems, err := sched.RunNow(ctx, cfg.ID) + stats, err := sched.RunNow(ctx, cfg.ID, &progress) result <- struct { - items int + stats *scheduler.RunStats err error - }{items: newItems, err: err} + }{stats: stats, err: err} }() // Wait for result @@ -217,10 +228,30 @@ func handleRun(ctx context.Context, sess ssh.Session, user *store.User, st *stor return } - if res.items == 0 { - println(sess, dimStyle.Render("No new items found.")) - } else { - println(sess, successStyle.Render(fmt.Sprintf("Sent %d new item(s) to %s", res.items, cfg.Email))) + // Display detailed stats + if res.stats != nil { + if res.stats.FailedFeeds > 0 { + printf(sess, "%s Fetched %d/%d feeds (%d failed)\n", + dimStyle.Render("⚠"), + res.stats.FetchedFeeds, + res.stats.TotalFeeds, + res.stats.FailedFeeds) + } else { + printf(sess, "%s Fetched %d/%d feeds\n", + successStyle.Render("✓"), + res.stats.FetchedFeeds, + res.stats.TotalFeeds) + } + + if res.stats.NewItems == 0 { + println(sess, dimStyle.Render("No new items found.")) + } else { + if res.stats.EmailSent { + println(sess, successStyle.Render(fmt.Sprintf("Sent %d new item(s) to %s", res.stats.NewItems, cfg.Email))) + } else { + println(sess, dimStyle.Render(fmt.Sprintf("Found %d new item(s) but did not send email", res.stats.NewItems))) + } + } } } diff --git a/store/tracking.go b/store/tracking.go index f8370f4..cfc039d 100644 --- a/store/tracking.go +++ b/store/tracking.go @@ -42,6 +42,17 @@ func (db *DB) RecordEmailSend(configID int64, recipient, subject string, include return trackingToken, nil } +// RecordEmailSendTx records an email send within an existing transaction +func (db *DB) RecordEmailSendTx(tx *sql.Tx, configID int64, recipient, subject, trackingToken string) error { + query := `INSERT INTO email_sends (config_id, recipient, subject, tracking_token) + VALUES (?, ?, ?, ?)` + _, err := tx.Exec(query, configID, recipient, subject, sql.NullString{String: trackingToken, Valid: trackingToken != ""}) + if err != nil { + return fmt.Errorf("insert email send: %w", err) + } + return nil +} + // MarkEmailBounced marks an email as bounced func (db *DB) MarkEmailBounced(configID int64, recipient, reason string) error { query := `UPDATE email_sends @@ -170,6 +181,15 @@ func (db *DB) CleanupOldSends(daysToKeep int) (int64, error) { return result.RowsAffected() } +// GenerateTrackingToken generates a secure random tracking token +func (db *DB) GenerateTrackingToken() (string, error) { + b := make([]byte, 24) + if _, err := rand.Read(b); err != nil { + return "", err + } + return base64.URLEncoding.EncodeToString(b), nil +} + func generateTrackingToken() (string, error) { b := make([]byte, 24) if _, err := rand.Read(b); err != nil { diff --git a/web/handlers.go b/web/handlers.go index c7e5e0f..e12a3fb 100644 --- a/web/handlers.go +++ b/web/handlers.go @@ -1,6 +1,7 @@ package web import ( + "context" "database/sql" "encoding/json" "encoding/xml" @@ -90,6 +91,9 @@ func (s *Server) handleUser(w http.ResponseWriter, r *http.Request, fingerprint s.handle404(w, r) return } + if errors.Is(err, context.Canceled) { + return // Client disconnected, don't log as error + } s.logger.Warn("get user", "err", err) http.Error(w, "Internal Server Error", http.StatusInternalServerError) return @@ -229,6 +233,9 @@ func (s *Server) handleFeedXML(w http.ResponseWriter, r *http.Request, fingerpri s.handle404(w, r) return } + if errors.Is(err, context.Canceled) { + return // Client disconnected + } s.logger.Warn("get user", "err", err) http.Error(w, "Internal Server Error", http.StatusInternalServerError) return @@ -358,6 +365,9 @@ func (s *Server) handleFeedJSON(w http.ResponseWriter, r *http.Request, fingerpr s.handle404(w, r) return } + if errors.Is(err, context.Canceled) { + return // Client disconnected + } s.logger.Warn("get user", "err", err) http.Error(w, "Internal Server Error", http.StatusInternalServerError) return @@ -464,6 +474,9 @@ func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request, fingerprin s.handle404(w, r) return } + if errors.Is(err, context.Canceled) { + return // Client disconnected + } s.logger.Warn("get user", "err", err) http.Error(w, "Internal Server Error", http.StatusInternalServerError) return