fix: harden schedulers, row iteration, and session handling

README.md

@@ -211,6 +211,7 @@ Build the binary, place it and the config, then install the service:
 
 ```bash
 go build -o /usr/local/bin/helm ./cmd/helm
+install -d /etc/helm
 cp config.yml /etc/helm/config.yml
 ```
 
@@ -225,13 +226,16 @@ ExecStart=/usr/local/bin/helm /etc/helm/config.yml
 WorkingDirectory=/var/lib/helm
 User=helm
 Group=helm
+Environment=TZ=UTC
 Restart=on-failure
 RestartSec=5s
 
 [Install]
 WantedBy=multi-user.target
 ```
 
+> `TZ` controls the timezone Helm uses for reminder/recurrence scheduling. Change `UTC` to your local tz (e.g. `America/New_York`) so scheduled events fire at the expected wall-clock time. `tzdata` must be installed on the host.
+
 ```bash
 useradd -r -s /sbin/nologin helm
 mkdir -p /var/lib/helm/data

cmd/helm/main.go

@@ -10,6 +10,7 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 	"time"
 
@@ -34,6 +35,8 @@ func main() {
 		log.Fatalf("config: %v", err)
 	}
 
+	log.Printf("timezone: %s (set TZ env var to override)", time.Local.String())
+
 	database, err := db.Open(cfg.Storage.DBPath)
 	if err != nil {
 		log.Fatalf("database: %v", err)
@@ -48,14 +51,17 @@ func main() {
 		log.Fatalf("attachments dir: %v", err)
 	}
 
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
 	b := broker.New()
-	stopReminders := reminder.StartScheduler(database, b)
+	stopReminders := reminder.StartScheduler(ctx, database, b)
 	defer stopReminders()
 
-	stopRecurrence := recurrence.StartScheduler(database)
+	stopRecurrence := recurrence.StartScheduler(ctx, database)
 	defer stopRecurrence()
 
-	stopCalDAV := startCalDAVScheduler(database, cfg.Auth.Secret)
+	stopCalDAV := startCalDAVScheduler(ctx, database, cfg.Auth.Secret)
 	defer stopCalDAV()
 
 	var uiFS fs.FS
@@ -70,9 +76,6 @@ func main() {
 	addr := fmt.Sprintf("%s:%d", cfg.Server.Host, cfg.Server.Port)
 	srv := &http.Server{Addr: addr, Handler: router}
 
-	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
-	defer stop()
-
 	go func() {
 		log.Printf("helm listening on http://%s", addr)
 		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
@@ -91,13 +94,14 @@ func main() {
 }
 
 // startCalDAVScheduler syncs all non-local calendar sources every 15 minutes.
-// Returns a cancel function.
-func startCalDAVScheduler(database *sql.DB, secret string) func() {
-	ticker := time.NewTicker(15 * time.Minute)
-	done := make(chan struct{})
+// The scheduler stops when parent ctx is cancelled or when the returned stop function is
+// invoked. Stop blocks until the ticker goroutine and any in-flight sync goroutines return.
+func startCalDAVScheduler(parent context.Context, database *sql.DB, secret string) func() {
+	ctx, cancel := context.WithCancel(parent)
+	var wg sync.WaitGroup
 
 	syncAll := func() {
-		rows, err := database.Query(
+		rows, err := database.QueryContext(ctx,
 			`SELECT id, name, url, username, password_enc FROM calendar_sources WHERE is_local = 0`,
 		)
 		if err != nil {
@@ -116,25 +120,36 @@ func startCalDAVScheduler(database *sql.DB, secret string) func() {
 			src.Username = username.String
 			src.PasswordEnc = passwordEnc.String
 
+			wg.Add(1)
 			go func(s caldav.CalendarSource) {
+				defer wg.Done()
 				if err := caldav.SyncSource(database, s, secret); err != nil {
 					log.Printf("caldav scheduler: source %d: %v", s.ID, err)
 				}
 			}(src)
 		}
+		if err := rows.Err(); err != nil {
+			log.Printf("caldav scheduler: iterate sources: %v", err)
+		}
 	}
 
+	wg.Add(1)
 	go func() {
+		defer wg.Done()
+		ticker := time.NewTicker(15 * time.Minute)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ticker.C:
 				syncAll()
-			case <-done:
-				ticker.Stop()
+			case <-ctx.Done():
 				return
 			}
 		}
 	}()
 
-	return func() { close(done) }
+	return func() {
+		cancel()
+		wg.Wait()
+	}
 }

internal/api/handlers/attachments.go

@@ -164,6 +164,10 @@ func ListAttachments(db *sql.DB) http.HandlerFunc {
 			}
 			attachments = append(attachments, a)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, attachments)
 	}
 }

internal/api/handlers/bookmarks.go

@@ -54,6 +54,10 @@ func ListBookmarkCollections(db *sql.DB) http.HandlerFunc {
 			}
 			collections = append(collections, c)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, collections)
 	}
 }
@@ -142,6 +146,10 @@ func ListBookmarks(db *sql.DB) http.HandlerFunc {
 			bookmarks = append(bookmarks, bm)
 			ids = append(ids, bm.ID)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		tagMap := batchGetEntityTags(db, "bookmark", ids)
 		for i := range bookmarks {
 			if tags, ok := tagMap[bookmarks[i].ID]; ok {

internal/api/handlers/calendar.go

@@ -65,6 +65,10 @@ func ListCalendarSources(db *sql.DB) http.HandlerFunc {
 			}
 			sources = append(sources, s)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, sources)
 	}
 }
@@ -221,6 +225,10 @@ func ListCalendarEvents(db *sql.DB) http.HandlerFunc {
 			}
 			events = append(events, e)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, events)
 	}
 }

internal/api/handlers/clipboard.go

@@ -60,6 +60,10 @@ func ListClipboardItems(db *sql.DB) http.HandlerFunc {
 			items = append(items, item)
 			ids = append(ids, item.ID)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		tagMap := batchGetEntityTags(db, "clipboard", ids)
 		for i := range items {
 			if tags, ok := tagMap[items[i].ID]; ok {

internal/api/handlers/memos.go

@@ -63,6 +63,10 @@ func ListMemos(db *sql.DB) http.HandlerFunc {
 			memos = append(memos, m)
 			ids = append(ids, m.ID)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		tagMap := batchGetEntityTags(db, "memo", ids)
 		for i := range memos {
 			if tags, ok := tagMap[memos[i].ID]; ok {

internal/api/handlers/notes.go

@@ -52,6 +52,10 @@ func ListNoteFolders(db *sql.DB) http.HandlerFunc {
 			}
 			folders = append(folders, f)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, folders)
 	}
 }
@@ -140,6 +144,10 @@ func ListNotes(db *sql.DB) http.HandlerFunc {
 			notes = append(notes, n)
 			ids = append(ids, n.ID)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		tagMap := batchGetEntityTags(db, "note", ids)
 		for i := range notes {
 			if tags, ok := tagMap[notes[i].ID]; ok {

internal/api/handlers/reminders.go

@@ -40,6 +40,10 @@ func ListReminders(db *sql.DB) http.HandlerFunc {
 			}
 			reminders = append(reminders, rem)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, reminders)
 	}
 }

internal/api/handlers/tags.go

@@ -65,6 +65,10 @@ func ListTags(db *sql.DB) http.HandlerFunc {
 			}
 			tags = append(tags, t)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, tags)
 	}
 }

internal/api/handlers/todos.go

@@ -77,6 +77,10 @@ func ListTodoLists(db *sql.DB) http.HandlerFunc {
 			}
 			lists = append(lists, l)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		respond(w, http.StatusOK, lists)
 	}
 }
@@ -180,6 +184,10 @@ func ListTodos(db *sql.DB) http.HandlerFunc {
 			todos = append(todos, t)
 			ids = append(ids, t.ID)
 		}
+		if err := rows.Err(); err != nil {
+			respondError(w, http.StatusInternalServerError, "row iteration failed")
+			return
+		}
 		tagMap := batchGetEntityTags(db, "todo", ids)
 		subtaskMap := batchGetSubtasks(db, ids)
 		for i := range todos {

internal/caldav/sync.go

@@ -2,14 +2,26 @@ package caldav
 
 import (
 	"database/sql"
+	"errors"
 	"fmt"
 	"log"
+	"sync"
 	"time"
 
 	"github.com/lerko/helm/internal/crypto"
 	"github.com/lerko/helm/internal/httpclient"
 )
 
+// ErrSyncInProgress is returned when a concurrent sync is already running for the same source.
+var ErrSyncInProgress = errors.New("sync already in progress for source")
+
+var syncLocks sync.Map // map[int64]*sync.Mutex
+
+func lockFor(id int64) *sync.Mutex {
+	v, _ := syncLocks.LoadOrStore(id, &sync.Mutex{})
+	return v.(*sync.Mutex)
+}
+
 // validateCalDAVURL defers to the shared SSRF policy: https-only + no
 // private/loopback destinations.
 func validateCalDAVURL(rawURL string) error {
@@ -27,7 +39,14 @@ type CalendarSource struct {
 
 // SyncSource fetches events from a remote CalDAV source and upserts them into the DB.
 // It skips events whose etag is unchanged, and deletes DB events not present in the remote response.
+// Returns ErrSyncInProgress if a concurrent sync for the same source is already running.
 func SyncSource(db *sql.DB, source CalendarSource, secret string) error {
+	mu := lockFor(source.ID)
+	if !mu.TryLock() {
+		return ErrSyncInProgress
+	}
+	defer mu.Unlock()
+
 	if err := validateCalDAVURL(source.URL); err != nil {
 		return fmt.Errorf("source %d URL rejected: %w", source.ID, err)
 	}
@@ -154,6 +173,9 @@ func deleteStaleEvents(db *sql.DB, sourceID int64, keepUIDs map[string]struct{})
 			toDelete = append(toDelete, uid)
 		}
 	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
 	rows.Close()
 
 	for _, uid := range toDelete {

internal/caldav/sync_test.go

@@ -0,0 +1,54 @@
+package caldav
+
+import (
+	"errors"
+	"strings"
+	"testing"
+)
+
+func TestSyncSource_ConcurrentCallsReturnErrSyncInProgress(t *testing.T) {
+	// Acquire the per-source lock manually to simulate an in-flight sync.
+	mu := lockFor(999)
+	mu.Lock()
+	defer mu.Unlock()
+
+	src := CalendarSource{
+		ID:   999,
+		Name: "concurrent-test",
+		URL:  "https://example.invalid/",
+	}
+
+	err := SyncSource(nil, src, "secret")
+	if !errors.Is(err, ErrSyncInProgress) {
+		t.Fatalf("expected ErrSyncInProgress, got %v", err)
+	}
+}
+
+func TestSyncSource_PrivateIPRejected(t *testing.T) {
+	// Fresh source ID so the lock is free and the URL check runs.
+	src := CalendarSource{
+		ID:   1001,
+		Name: "ssrf-test",
+		URL:  "http://127.0.0.1/",
+	}
+
+	err := SyncSource(nil, src, "secret")
+	if err == nil {
+		t.Fatal("expected SSRF rejection, got nil")
+	}
+	if !strings.Contains(err.Error(), "rejected") {
+		t.Errorf("expected rejection error, got: %v", err)
+	}
+}
+
+func TestSyncSource_MetadataEndpointRejected(t *testing.T) {
+	src := CalendarSource{
+		ID:   1002,
+		Name: "aws-metadata",
+		URL:  "https://169.254.169.254/latest/meta-data/",
+	}
+	err := SyncSource(nil, src, "secret")
+	if err == nil {
+		t.Fatal("expected SSRF rejection, got nil")
+	}
+}