feat: security / reliability fixes

scheduler/fetch.go

···
 func FetchFeeds(ctx context.Context, feeds []*store.Feed) []*FetchResult {
 	results := make([]*FetchResult, len(feeds))
 	var wg sync.WaitGroup
+	
+	// Limit concurrent fetches to 10 to avoid overwhelming the system
+	maxConcurrent := 10
+	if len(feeds) < maxConcurrent {
+		maxConcurrent = len(feeds)
+	}
+	semaphore := make(chan struct{}, maxConcurrent)
 
 	for i, feed := range feeds {
 		wg.Add(1)
 		go func(idx int, f *store.Feed) {
 			defer wg.Done()
+			semaphore <- struct{}{}        // Acquire
+			defer func() { <-semaphore }() // Release
 			results[idx] = FetchFeed(ctx, f)
 		}(i, feed)
 	}

scheduler/scheduler.go

···
 
 	results := FetchFeeds(ctx, feeds)
 
+	feedGroups, totalNew, err := s.collectNewItems(ctx, results)
+	if err != nil {
+		return 0, err
+	}
+
+	if totalNew > 0 {
+		if err := s.sendDigestAndMarkSeen(ctx, cfg, feedGroups, totalNew, results); err != nil {
+			return 0, err
+		}
+		s.logger.Info("email sent", "to", cfg.Email, "items", totalNew)
+	}
+
+	// Update feed metadata
+	for _, result := range results {
+		if result.ETag != "" || result.LastModified != "" {
+			if err := s.store.UpdateFeedFetched(ctx, result.FeedID, result.ETag, result.LastModified); err != nil {
+				s.logger.Warn("failed to update feed fetched", "err", err)
+			}
+		}
+	}
+
+	now := time.Now()
+	nextRun, err := gronx.NextTick(cfg.CronExpr, false)
+	if err != nil {
+		return totalNew, fmt.Errorf("calculate next run: %w", err)
+	}
+
+	if err := s.store.UpdateLastRun(ctx, cfg.ID, now, nextRun); err != nil {
+		return totalNew, 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
+}
+
+func (s *Scheduler) collectNewItems(ctx context.Context, results []*FetchResult) ([]email.FeedGroup, int, error) {
 	var feedGroups []email.FeedGroup
 	totalNew := 0
 	threeMonthsAgo := time.Now().AddDate(0, -3, 0)
···
 			})
 			totalNew += len(newItems)
 		}
-
-		if result.ETag != "" || result.LastModified != "" {
-			if err := s.store.UpdateFeedFetched(ctx, result.FeedID, result.ETag, result.LastModified); err != nil {
-				s.logger.Warn("failed to update feed fetched", "err", err)
-			}
-		}
 	}
 
 	if feedErrors == len(results) {
-		return 0, fmt.Errorf("all feeds failed to fetch")
+		return nil, 0, fmt.Errorf("all feeds failed to fetch")
 	}
 
-	if totalNew > 0 {
-		digestData := &email.DigestData{
-			ConfigName: cfg.Filename,
-			TotalItems: totalNew,
-			FeedGroups: feedGroups,
-		}
+	return feedGroups, totalNew, nil
+}
 
-		inline := cfg.InlineContent
-		if totalNew > 5 {
-			inline = false
-		}
+func (s *Scheduler) sendDigestAndMarkSeen(ctx context.Context, cfg *store.Config, feedGroups []email.FeedGroup, totalNew int, results []*FetchResult) error {
+	digestData := &email.DigestData{
+		ConfigName: cfg.Filename,
+		TotalItems: totalNew,
+		FeedGroups: feedGroups,
+	}
 
-		htmlBody, textBody, err := email.RenderDigest(digestData, inline)
-		if err != nil {
-			return 0, fmt.Errorf("render digest: %w", err)
-		}
+	inline := cfg.InlineContent
+	if totalNew > 5 {
+		inline = false
+	}
 
-		unsubToken, err := s.store.GetOrCreateUnsubscribeToken(ctx, cfg.ID)
-		if err != nil {
-			s.logger.Warn("failed to create unsubscribe token", "err", err)
-			unsubToken = ""
-		}
+	htmlBody, textBody, err := email.RenderDigest(digestData, inline)
+	if err != nil {
+		return fmt.Errorf("render digest: %w", err)
+	}
 
-		// Get user fingerprint for dashboard URL
-		user, err := s.store.GetUserByID(ctx, cfg.UserID)
-		dashboardURL := ""
-		if err == nil {
-			dashboardURL = s.originURL + "/" + user.PubkeyFP
-		} else {
-			s.logger.Warn("failed to get user for dashboard URL", "err", err)
-		}
+	unsubToken, err := s.store.GetOrCreateUnsubscribeToken(ctx, cfg.ID)
+	if err != nil {
+		s.logger.Warn("failed to create unsubscribe token", "err", err)
+		unsubToken = ""
+	}
 
-		subject := "feed digest"
-		if err := s.mailer.Send(cfg.Email, subject, htmlBody, textBody, unsubToken, dashboardURL); err != nil {
-			return 0, fmt.Errorf("send email: %w", err)
-		}
+	user, err := s.store.GetUserByID(ctx, cfg.UserID)
+	dashboardURL := ""
+	if err == nil {
+		dashboardURL = s.originURL + "/" + user.PubkeyFP
+	} else {
+		s.logger.Warn("failed to get user for dashboard URL", "err", err)
+	}
 
-		s.logger.Info("email sent", "to", cfg.Email, "items", totalNew)
+	// Begin transaction to mark items seen
+	tx, err := s.store.BeginTx(ctx)
+	if err != nil {
+		return fmt.Errorf("begin transaction: %w", err)
+	}
+	defer tx.Rollback()
 
-		for _, result := range results {
-			if result.Error != nil {
-				continue
-			}
-			for _, item := range result.Items {
-				if err := s.store.MarkItemSeen(ctx, result.FeedID, item.GUID, item.Title, item.Link); err != nil {
-					s.logger.Warn("failed to mark item seen", "err", err)
-				}
+	// Mark items seen BEFORE sending email
+	for _, result := range results {
+		if result.Error != nil {
+			continue
+		}
+		for _, item := range result.Items {
+			if err := s.store.MarkItemSeenTx(ctx, tx, result.FeedID, item.GUID, item.Title, item.Link); err != nil {
+				s.logger.Warn("failed to mark item seen", "err", err)
 			}
 		}
 	}
 
-	now := time.Now()
-	nextRun, err := gronx.NextTick(cfg.CronExpr, false)
-	if err != nil {
-		return totalNew, fmt.Errorf("calculate next run: %w", err)
+	// Send email - if this fails, transaction will rollback
+	subject := "feed digest"
+	if err := s.mailer.Send(cfg.Email, subject, htmlBody, textBody, unsubToken, dashboardURL); err != nil {
+		return fmt.Errorf("send email: %w", err)
 	}
 
-	if err := s.store.UpdateLastRun(ctx, cfg.ID, now, nextRun); err != nil {
-		return totalNew, fmt.Errorf("update last run: %w", err)
+	// Commit transaction only after successful email send
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("commit transaction: %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
+	return nil
 }
 
 func (s *Scheduler) processConfig(ctx context.Context, cfg *store.Config) error {
···
 
 	results := FetchFeeds(ctx, feeds)
 
-	var feedGroups []email.FeedGroup
-	totalNew := 0
-	threeMonthsAgo := time.Now().AddDate(0, -3, 0)
-
-	for _, result := range results {
-		if result.Error != nil {
-			s.logger.Warn("feed fetch error", "feed_id", result.FeedID, "url", result.FeedURL, "err", result.Error)
-			continue
-		}
-
-		var newItems []email.FeedItem
-		for _, item := range result.Items {
-			// Skip items older than 3 months
-			if !item.Published.IsZero() && item.Published.Before(threeMonthsAgo) {
-				continue
-			}
-
-			seen, err := s.store.IsItemSeen(ctx, result.FeedID, item.GUID)
-			if err != nil {
-				s.logger.Warn("failed to check if item seen", "err", err)
-				continue
-			}
+	feedGroups, totalNew, err := s.collectNewItems(ctx, results)
+	if err != nil {
+		s.logger.Warn("failed to collect items", "config_id", cfg.ID, "err", err)
+	}
 
-			if !seen {
-				newItems = append(newItems, email.FeedItem{
-					Title:     item.Title,
-					Link:      item.Link,
-					Content:   item.Content,
-					Published: item.Published,
-				})
-			}
+	if totalNew > 0 {
+		if err := s.sendDigestAndMarkSeen(ctx, cfg, feedGroups, totalNew, results); err != nil {
+			return fmt.Errorf("send digest: %w", err)
 		}
+		s.logger.Info("email sent", "to", cfg.Email, "items", totalNew)
+	} else {
+		s.logger.Info("no new items", "config_id", cfg.ID)
+	}
 
-		if len(newItems) > 0 {
-			feedName := result.FeedName
-			if feedName == "" {
-				feedName = result.FeedURL
-			}
-			feedGroups = append(feedGroups, email.FeedGroup{
-				FeedName: feedName,
-				FeedURL:  result.FeedURL,
-				Items:    newItems,
-			})
-			totalNew += len(newItems)
-		}
-
+	// Update feed metadata
+	for _, result := range results {
 		if result.ETag != "" || result.LastModified != "" {
 			if err := s.store.UpdateFeedFetched(ctx, result.FeedID, result.ETag, result.LastModified); err != nil {
 				s.logger.Warn("failed to update feed fetched", "err", err)
-			}
-		}
-	}
-
-	if totalNew == 0 {
-		s.logger.Info("no new items", "config_id", cfg.ID)
-	} else {
-		digestData := &email.DigestData{
-			ConfigName: cfg.Filename,
-			TotalItems: totalNew,
-			FeedGroups: feedGroups,
-		}
-
-		// Auto-disable inline content if more than 5 items
-		inline := cfg.InlineContent
-		if totalNew > 5 {
-			inline = false
-		}
-
-		htmlBody, textBody, err := email.RenderDigest(digestData, inline)
-		if err != nil {
-			return fmt.Errorf("render digest: %w", err)
-		}
-
-		unsubToken, err := s.store.GetOrCreateUnsubscribeToken(ctx, cfg.ID)
-		if err != nil {
-			s.logger.Warn("failed to create unsubscribe token", "err", err)
-			unsubToken = ""
-		}
-
-		// Get user fingerprint for dashboard URL
-		user, err := s.store.GetUserByID(ctx, cfg.UserID)
-		dashboardURL := ""
-		if err == nil {
-			dashboardURL = s.originURL + "/" + user.PubkeyFP
-		} else {
-			s.logger.Warn("failed to get user for dashboard URL", "err", err)
-		}
-
-		subject := "feed digest"
-		if err := s.mailer.Send(cfg.Email, subject, htmlBody, textBody, unsubToken, dashboardURL); err != nil {
-			return fmt.Errorf("send email: %w", err)
-		}
-
-		s.logger.Info("email sent", "to", cfg.Email, "items", totalNew)
-
-		for _, result := range results {
-			if result.Error != nil {
-				continue
-			}
-			for _, item := range result.Items {
-				if err := s.store.MarkItemSeen(ctx, result.FeedID, item.GUID, item.Title, item.Link); err != nil {
-					s.logger.Warn("failed to mark item seen", "err", err)
-				}
 			}
 		}
 	}

ssh/scp.go

···
 	}
 
 	ctx := s.Context()
-	if err := h.store.DeleteConfig(ctx, user.ID, name); err != nil {
+	
+	// Use transaction for config update
+	tx, err := h.store.BeginTx(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("begin transaction: %w", err)
+	}
+	defer tx.Rollback()
+
+	if err := h.store.DeleteConfigTx(ctx, tx, user.ID, name); err != nil {
 		h.logger.Debug("no existing config to delete", "filename", name)
 	} else {
 		h.logger.Debug("deleted existing config", "filename", name)
 	}
 
-	cfg, err := h.store.CreateConfig(ctx, user.ID, name, parsed.Email, parsed.CronExpr, parsed.Digest, parsed.Inline, string(content), nextRun)
+	cfg, err := h.store.CreateConfigTx(ctx, tx, user.ID, name, parsed.Email, parsed.CronExpr, parsed.Digest, parsed.Inline, string(content), nextRun)
 	if err != nil {
 		return 0, fmt.Errorf("failed to save config: %w", err)
 	}
 
 	for _, feed := range parsed.Feeds {
-		if _, err := h.store.CreateFeed(ctx, cfg.ID, feed.URL, feed.Name); err != nil {
+		if _, err := h.store.CreateFeedTx(ctx, tx, cfg.ID, feed.URL, feed.Name); err != nil {
 			return 0, fmt.Errorf("failed to save feed: %w", err)
 		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return 0, fmt.Errorf("commit transaction: %w", err)
 	}
 
 	h.logger.Info("config uploaded", "user_id", user.ID, "filename", name, "feeds", len(parsed.Feeds), "next_run", nextRun)

store/configs.go

···
 	}, nil
 }
 
+func (db *DB) CreateConfigTx(ctx context.Context, tx *sql.Tx, userID int64, filename, email, cronExpr string, digest, inline bool, rawText string, nextRun time.Time) (*Config, error) {
+	result, err := tx.ExecContext(ctx,
+		`INSERT INTO configs (user_id, filename, email, cron_expr, digest, inline_content, raw_text, next_run)
+		 VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+		userID, filename, email, cronExpr, digest, inline, rawText, nextRun,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("insert config: %w", err)
+	}
+
+	id, err := result.LastInsertId()
+	if err != nil {
+		return nil, fmt.Errorf("get last insert id: %w", err)
+	}
+
+	return &Config{
+		ID:            id,
+		UserID:        userID,
+		Filename:      filename,
+		Email:         email,
+		CronExpr:      cronExpr,
+		Digest:        digest,
+		InlineContent: inline,
+		RawText:       rawText,
+		NextRun:       sql.NullTime{Time: nextRun, Valid: true},
+		CreatedAt:     time.Now(),
+	}, nil
+}
+
+func (db *DB) DeleteConfigTx(ctx context.Context, tx *sql.Tx, userID int64, filename string) error {
+	result, err := tx.ExecContext(ctx,
+		`DELETE FROM configs WHERE user_id = ? AND filename = ?`,
+		userID, filename,
+	)
+	if err != nil {
+		return fmt.Errorf("delete config: %w", err)
+	}
+
+	n, err := result.RowsAffected()
+	if err != nil {
+		return fmt.Errorf("rows affected: %w", err)
+	}
+	if n == 0 {
+		return sql.ErrNoRows
+	}
+	return nil
+}
+
 func (db *DB) GetConfig(ctx context.Context, userID int64, filename string) (*Config, error) {
 	var cfg Config
 	err := db.QueryRowContext(ctx,

store/db.go

···
 }
 
 func Open(path string) (*DB, error) {
-	db, err := sql.Open("sqlite3", path+"?_foreign_keys=on")
+	db, err := sql.Open("sqlite3", path+"?_foreign_keys=on&_journal_mode=WAL&_busy_timeout=5000")
 	if err != nil {
 		return nil, fmt.Errorf("open database: %w", err)
 	}
+
+	// SQLite works best with single writer for WAL mode
+	db.SetMaxOpenConns(1)
+	db.SetMaxIdleConns(1)
 
 	if err := db.Ping(); err != nil {
 		return nil, fmt.Errorf("ping database: %w", err)
···
 	);
 
 	CREATE INDEX IF NOT EXISTS idx_configs_user_id ON configs(user_id);
-	CREATE INDEX IF NOT EXISTS idx_configs_next_run ON configs(next_run);
+	CREATE INDEX IF NOT EXISTS idx_configs_active_next_run ON configs(next_run) WHERE next_run IS NOT NULL;
 	CREATE INDEX IF NOT EXISTS idx_feeds_config_id ON feeds(config_id);
 	CREATE INDEX IF NOT EXISTS idx_seen_items_feed_id ON seen_items(feed_id);
 	CREATE INDEX IF NOT EXISTS idx_logs_config_id ON logs(config_id);

store/feeds.go

···
 	}, nil
 }
 
+func (db *DB) CreateFeedTx(ctx context.Context, tx *sql.Tx, configID int64, url, name string) (*Feed, error) {
+	var nameVal sql.NullString
+	if name != "" {
+		nameVal = sql.NullString{String: name, Valid: true}
+	}
+
+	result, err := tx.ExecContext(ctx,
+		`INSERT INTO feeds (config_id, url, name) VALUES (?, ?, ?)`,
+		configID, url, nameVal,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("insert feed: %w", err)
+	}
+
+	id, err := result.LastInsertId()
+	if err != nil {
+		return nil, fmt.Errorf("get last insert id: %w", err)
+	}
+
+	return &Feed{
+		ID:       id,
+		ConfigID: configID,
+		URL:      url,
+		Name:     nameVal,
+	}, nil
+}
+
 func (db *DB) GetFeedsByConfig(ctx context.Context, configID int64) ([]*Feed, error) {
 	rows, err := db.QueryContext(ctx,
 		`SELECT id, config_id, url, name, last_fetched, etag, last_modified

store/items.go

···
 	return true, nil
 }
 
+func (db *DB) MarkItemSeenTx(ctx context.Context, tx *sql.Tx, feedID int64, guid, title, link string) error {
+	var titleVal, linkVal sql.NullString
+	if title != "" {
+		titleVal = sql.NullString{String: title, Valid: true}
+	}
+	if link != "" {
+		linkVal = sql.NullString{String: link, Valid: true}
+	}
+
+	_, err := tx.ExecContext(ctx,
+		`INSERT INTO seen_items (feed_id, guid, title, link) VALUES (?, ?, ?, ?)
+		 ON CONFLICT(feed_id, guid) DO UPDATE SET title = excluded.title, link = excluded.link`,
+		feedID, guid, titleVal, linkVal,
+	)
+	if err != nil {
+		return fmt.Errorf("mark item seen: %w", err)
+	}
+	return nil
+}
+
 func (db *DB) GetSeenItems(ctx context.Context, feedID int64, limit int) ([]*SeenItem, error) {
 	rows, err := db.QueryContext(ctx,
 		`SELECT id, feed_id, guid, title, link, seen_at