initial commit on Federted servers

internal/signals/checkpoint_test.go

@@ -0,0 +1,347 @@
+package signals
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/emersion/go-imap"
+	"github.com/emersion/go-imap/backend"
+	"github.com/emersion/go-imap/backend/memory"
+	imapserver "github.com/emersion/go-imap/server"
+)
+
+// validityBackend wraps the in-memory backend so tests can force the
+// UIDVALIDITY the server reports (the stock memory backend hardcodes 1) and
+// capture the SEARCH criteria every pass issues (review CR-15/CR-19
+// verification).
+type validityBackend struct {
+	backend.Backend
+	validity uint32
+	mu       sync.Mutex
+	criteria []*imap.SearchCriteria
+	// failFetchAfter makes the fixture fail ListMessages calls after the
+	// given count, simulating a mailbox that dies partway through a pass.
+	failFetchAfter int
+	fetches        int
+}
+
+func (b *validityBackend) Login(connInfo *imap.ConnInfo, username, password string) (backend.User, error) {
+	u, e := b.Backend.Login(connInfo, username, password)
+	if e != nil {
+		return nil, e
+	}
+	return &validityUser{User: u, be: b}, nil
+}
+
+func (b *validityBackend) captured() []*imap.SearchCriteria {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return append([]*imap.SearchCriteria(nil), b.criteria...)
+}
+
+type validityUser struct {
+	backend.User
+	be *validityBackend
+}
+
+func (u *validityUser) GetMailbox(name string) (backend.Mailbox, error) {
+	m, e := u.User.GetMailbox(name)
+	if e != nil {
+		return nil, e
+	}
+	return &validityMailbox{Mailbox: m, be: u.be}, nil
+}
+
+type validityMailbox struct {
+	backend.Mailbox
+	be *validityBackend
+}
+
+func (m *validityMailbox) Status(items []imap.StatusItem) (*imap.MailboxStatus, error) {
+	st, e := m.Mailbox.Status(items)
+	if e == nil && m.be.validity != 0 {
+		st.UidValidity = m.be.validity
+	}
+	return st, e
+}
+
+func (m *validityMailbox) SearchMessages(uid bool, c *imap.SearchCriteria) ([]uint32, error) {
+	if uid {
+		m.be.mu.Lock()
+		m.be.criteria = append(m.be.criteria, c)
+		m.be.mu.Unlock()
+	}
+	return m.Mailbox.SearchMessages(uid, c)
+}
+
+func (m *validityMailbox) ListMessages(uid bool, seqset *imap.SeqSet, items []imap.FetchItem, ch chan<- *imap.Message) error {
+	m.be.mu.Lock()
+	m.be.fetches++
+	fail := m.be.failFetchAfter > 0 && m.be.fetches > m.be.failFetchAfter
+	m.be.mu.Unlock()
+	if fail {
+		close(ch)
+		return errors.New("fixture fetch failure")
+	}
+	return m.Mailbox.ListMessages(uid, seqset, items, ch)
+}
+
+// newTestMailbox stands up the in-memory IMAP fixture the integration suite
+// uses, reseeded with n messages in INBOX (the backend ships one). The server
+// enforces real protocol semantics, so SearchConnFrom exercises
+// UidSearch/UidFetch against genuine UIDVALIDITY/UIDNEXT responses.
+func newTestMailbox(t *testing.T, n int) (Mailbox, func()) {
+	m, _, cleanup := newTestMailboxValidity(t, n, 0)
+	return m, cleanup
+}
+
+// newTestMailboxValidity is newTestMailbox with a forced UIDVALIDITY (0 keeps
+// the backend default) and access to the criteria capture.
+func newTestMailboxValidity(t *testing.T, n int, validity uint32) (Mailbox, *validityBackend, func()) {
+	t.Helper()
+	mb := memory.New()
+	// Reach the fixture's own user/mailbox through a refused login, which is
+	// how the backend exposes them.
+	_, err := mb.Login(nil, "username", "bad")
+	if err == nil {
+		t.Fatal("fixture login unexpectedly succeeded")
+	}
+	user, err := mb.Login(nil, "username", "password")
+	if err != nil {
+		t.Fatal(err)
+	}
+	mbox, err := user.GetMailbox("INBOX")
+	if err != nil {
+		t.Fatal(err)
+	}
+	memMbox := mbox.(*memory.Mailbox)
+	memMbox.Messages = memMbox.Messages[:0]
+	for i := 0; i < n; i++ {
+		body := "From: sender@computer.test\r\n" +
+			"Subject: subject " + fmt.Sprint(i) + "\r\n" +
+			"Message-Id: <msg-" + fmt.Sprint(i) + "@computer.test>\r\n" +
+			"\r\n" + "body"
+		memMbox.Messages = append(memMbox.Messages, &memory.Message{
+			Uid:  uint32(i + 1),
+			Date: time.Now(),
+			Size: uint32(len(body)),
+			Body: []byte(body),
+		})
+	}
+	listener, e := net.Listen("tcp", "127.0.0.1:0")
+	if e != nil {
+		t.Fatal(e)
+	}
+	be := &validityBackend{Backend: mb, validity: validity}
+	mail := imapserver.New(be)
+	mail.AllowInsecureAuth = true
+	go mail.Serve(listener)
+	host, port, _ := net.SplitHostPort(listener.Addr().String())
+	var p int
+	fmt.Sscanf(port, "%d", &p)
+	m := Mailbox{Host: host, Username: "username", Password: "password", Port: p, TLS: false, Folder: "INBOX"}
+	t.Setenv("INTEGRATIONS_ALLOW_INSECURE_IMAP", "1")
+	return m, be, func() { mail.Close() }
+}
+
+// TestTruncatedPassKeepsRemainingMatches pins review COR-1: when the
+// per-pass cap truncates the matched UID list, the checkpoint must advance
+// only to the last fetched UID — never to the mailbox high-water mark — or
+// matches past the cap are dropped forever.
+func TestTruncatedPassKeepsRemainingMatches(t *testing.T) {
+	// Connect dials loopback through netguard; the fixture flag unblocks it
+	// (same escape hatch the integration harness uses).
+	t.Setenv("INTEGRATIONS_ALLOW_PRIVATE", "1")
+	cap := searchBatch * maxIngestBatches
+	m, cleanup := newTestMailbox(t, cap+7)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	c, e := Connect(ctx, m)
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer c.Logout()
+
+	events, next, e := SearchConnFrom(c, m, "", "subject", Checkpoint{})
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(events) != cap {
+		t.Fatalf("fetched %d events, want %d (per-pass cap)", len(events), cap)
+	}
+	if next.LastUID != uint32(cap) {
+		t.Fatalf("checkpoint advanced to %d, want %d (high-water shortcut must stay off after truncation)", next.LastUID, cap)
+	}
+	// A follow-up pass from the new checkpoint picks up every remaining match.
+	rest, final, e := SearchConnFrom(c, m, "", "subject", next)
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(rest) != 7 {
+		t.Fatalf("second pass fetched %d events, want 7", len(rest))
+	}
+	if final.LastUID != uint32(cap+7) {
+		t.Fatalf("final checkpoint %d, want %d", final.LastUID, cap+7)
+	}
+}
+
+// Review follow-up: a mailbox that fails after an earlier batch succeeded
+// still returns the events already fetched. The caller persists them but must
+// not advance the checkpoint past the failed batch.
+func TestPartialFetchFailureReturnsEarlierEvents(t *testing.T) {
+	t.Setenv("INTEGRATIONS_ALLOW_PRIVATE", "1")
+	m, be, cleanup := newTestMailboxValidity(t, searchBatch*2+5, 0)
+	defer cleanup()
+	be.mu.Lock()
+	be.failFetchAfter = 1
+	be.mu.Unlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	c, e := Connect(ctx, m)
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer c.Logout()
+
+	events, next, e := SearchConnFrom(c, m, "", "subject", Checkpoint{})
+	if e == nil {
+		t.Fatal("fixture fetch failure was not reported")
+	}
+	if len(events) != searchBatch {
+		t.Fatalf("partial pass returned %d events, want the first batch (%d)", len(events), searchBatch)
+	}
+	if next.LastUID != searchBatch {
+		t.Fatalf("partial checkpoint %d, want %d (never past the failed batch)", next.LastUID, searchBatch)
+	}
+}
+
+func TestCheckpointHighWaterShortcutAppliesWhenComplete(t *testing.T) {
+	t.Setenv("INTEGRATIONS_ALLOW_PRIVATE", "1")
+	// When the SEARCH considered every message (no truncation), the
+	// checkpoint may advance to the mailbox high-water mark.
+	m, cleanup := newTestMailbox(t, searchBatch+1)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	c, e := Connect(ctx, m)
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer c.Logout()
+
+	events, next, e := SearchConnFrom(c, m, "", "subject", Checkpoint{})
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(events) != searchBatch+1 {
+		t.Fatalf("fetched %d events, want %d", len(events), searchBatch+1)
+	}
+	if next.LastUID != uint32(searchBatch+1) {
+		t.Fatalf("checkpoint %d, want %d", next.LastUID, searchBatch+1)
+	}
+}
+
+// TestUIDValidityResetReingests pins review CR-15 with a fixture-backed
+// ingestion pass: when UIDVALIDITY changes, the new UID space is scanned from
+// zero — even when every new UID sits BELOW the old checkpoint's LastUID.
+// The old behavior filtered against the stale high-water mark and skipped the
+// whole new space while advancing the checkpoint past it.
+func TestUIDValidityResetReingests(t *testing.T) {
+	t.Setenv("INTEGRATIONS_ALLOW_PRIVATE", "1")
+	// New UID space: UIDs 1..20 under UIDVALIDITY 8.
+	m, _, cleanup := newTestMailboxValidity(t, 20, 8)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	c, e := Connect(ctx, m)
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer c.Logout()
+
+	// A checkpoint from the PREVIOUS UID space whose LastUID (100) is above
+	// every UID that now exists.
+	stale := Checkpoint{UIDValidity: 7, LastUID: 100}
+	events, next, e := SearchConnFrom(c, m, "", "subject", stale)
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(events) != 20 {
+		t.Fatalf("reset pass ingested %d events, want all 20 of the new UID space", len(events))
+	}
+	if next.UIDValidity != 8 || next.LastUID != 20 {
+		t.Fatalf("checkpoint after reset = %+v, want {UIDValidity:8 LastUID:20}", next)
+	}
+	// A subsequent pass from the reset checkpoint finds nothing new and does
+	// not regress the checkpoint.
+	again, final, e := SearchConnFrom(c, m, "", "subject", next)
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(again) != 0 {
+		t.Fatalf("follow-up pass ingested %d events, want 0", len(again))
+	}
+	if final.UIDValidity != 8 || final.LastUID != 20 {
+		t.Fatalf("follow-up checkpoint = %+v, want {UIDValidity:8 LastUID:20}", final)
+	}
+}
+
+// TestIncrementalPassConstrainsUIDSearch pins review CR-19: once a source is
+// caught up, the server-side SEARCH carries a UID window derived from the
+// normalized checkpoint instead of re-matching the whole history. The reset
+// pass (CR-15) must NOT constrain — the whole new UID space is fair game.
+func TestIncrementalPassConstrainsUIDSearch(t *testing.T) {
+	t.Setenv("INTEGRATIONS_ALLOW_PRIVATE", "1")
+	m, be, cleanup := newTestMailboxValidity(t, 10, 5)
+	defer cleanup()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	c, e := Connect(ctx, m)
+	if e != nil {
+		t.Fatal(e)
+	}
+	defer c.Logout()
+
+	// Pass 1: mismatched UIDVALIDITY forces a reset; no UID constraint.
+	if _, next, e := SearchConnFrom(c, m, "", "subject", Checkpoint{UIDValidity: 4, LastUID: 99}); e != nil {
+		t.Fatal(e)
+	} else if next.LastUID != 10 {
+		t.Fatalf("reset pass checkpoint %d, want 10", next.LastUID)
+	}
+	// Pass 2: caught up at UID 4 of validity 5 — only UIDs 5..10 ingest.
+	events, next2, e := SearchConnFrom(c, m, "", "subject", Checkpoint{UIDValidity: 5, LastUID: 4})
+	if e != nil {
+		t.Fatal(e)
+	}
+	if len(events) != 6 {
+		t.Fatalf("incremental pass ingested %d events, want 6", len(events))
+	}
+	if next2.LastUID != 10 {
+		t.Fatalf("incremental checkpoint %d, want 10", next2.LastUID)
+	}
+
+	got := be.captured()
+	if len(got) != 2 {
+		t.Fatalf("captured %d UID searches, want 2", len(got))
+	}
+	if got[0].Uid != nil {
+		t.Fatalf("reset pass constrained the SEARCH to %v; it must rescan the whole UID space", got[0].Uid)
+	}
+	if got[1].Uid == nil {
+		t.Fatal("incremental pass did not constrain the SEARCH by UID window")
+	}
+	if !got[1].Uid.Contains(5) || got[1].Uid.Contains(4) {
+		t.Fatalf("incremental SEARCH UID window %v does not start at 5", got[1].Uid)
+	}
+}

internal/signals/scheduler.go

@@ -0,0 +1,153 @@
+package signals
+
+import (
+	"context"
+	"github.com/andrew/go-computer/internal/db"
+	"log"
+	"sync"
+	"time"
+)
+
+const (
+	// pollConcurrency bounds simultaneous mailbox passes; each pass holds one
+	// IMAP login and a 45-second budget (review SCALE-01).
+	pollConcurrency = 3
+	perUserTimeout  = 45 * time.Second
+	// maxBackoffShift caps failure backoff at 2^4 ticks (~80 minutes).
+	maxBackoffShift = 4
+)
+
+// Scheduler polls users with enabled signal sources. Review SCALE-01:
+// bounded concurrency, per-user exponential backoff on consecutive failures,
+// logged outcomes (counts only — never credentials or mail content), and a
+// database-backed advisory lease so multiple app instances never poll the
+// same user simultaneously. Database deduplication would hide duplicate
+// records but not the duplicated external work the lease prevents.
+type Scheduler struct {
+	mu       sync.Mutex
+	failures map[string]int
+	tick     int
+}
+
+func NewScheduler() *Scheduler { return &Scheduler{failures: map[string]int{}} }
+
+// RunOnce performs a single scheduler pass over every user with enabled
+// sources. Extracted from RunLoop so tests can drive one tick directly.
+func RunOnce(ctx context.Context, d *db.DB) { NewScheduler().RunOnce(ctx, d) }
+
+func (s *Scheduler) RunOnce(ctx context.Context, d *db.DB) {
+	users, e := db.Many(ctx, d, `SELECT jsonb_build_object('id',"ownerId") FROM "SignalSource" WHERE enabled GROUP BY "ownerId"`)
+	if e != nil {
+		log.Printf("watchers: query failed: %v", e)
+		return
+	}
+	s.mu.Lock()
+	s.tick++
+	tick := s.tick
+	s.mu.Unlock()
+	sem := make(chan struct{}, pollConcurrency)
+	var wg sync.WaitGroup
+	for _, u := range users {
+		if ctx.Err() != nil {
+			break
+		}
+		id, _ := u["id"].(string)
+		if id == "" || s.skipTick(id, tick) {
+			continue
+		}
+		wg.Add(1)
+		sem <- struct{}{}
+		go func() {
+			defer wg.Done()
+			defer func() { <-sem }()
+			s.checkUser(ctx, d, id)
+		}()
+	}
+	wg.Wait()
+}
+
+// skipTick applies per-user exponential backoff: after n consecutive failed
+// passes the user is polled on every 2^n-th tick instead of every tick.
+func (s *Scheduler) skipTick(id string, tick int) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	f := s.failures[id]
+	if f == 0 {
+		return false
+	}
+	shift := min(f, maxBackoffShift)
+	return tick%(1<<shift) != 0
+}
+
+func (s *Scheduler) checkUser(ctx context.Context, d *db.DB, id string) {
+	// Database-backed lease held on a dedicated connection for the duration
+	// of the pass: a second poller instance skips this user instead of
+	// repeating the same external mailbox work.
+	conn, e := d.Conn(ctx)
+	if e != nil {
+		s.note(id, false)
+		log.Printf("watchers: lease connection for user %s failed: %v", id, e)
+		return
+	}
+	defer conn.Close()
+	var locked bool
+	if e := conn.QueryRowContext(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,0))`, "signal-poll:"+id).Scan(&locked); e != nil {
+		s.note(id, false)
+		log.Printf("watchers: lease for user %s failed: %v", id, e)
+		return
+	}
+	if !locked {
+		return // another instance owns this user's pass
+	}
+	defer conn.ExecContext(context.WithoutCancel(ctx), `SELECT pg_advisory_unlock(hashtextextended($1,0))`, "signal-poll:"+id)
+	job, cancel := context.WithTimeout(ctx, perUserTimeout)
+	defer cancel()
+	res, e := RunSignalChecks(job, d, id, "")
+	if e != nil {
+		s.note(id, false)
+		log.Printf("watchers: check failed for user %s: %v", id, e)
+		return
+	}
+	if res.Failed > 0 {
+		// Review CR-16: real integration failures (mailbox auth/connection,
+		// search, insert, checkpoint or status writes) must drive backoff —
+		// previously they were recorded per source only and the pass counted
+		// as scheduler success, retrying every tick forever.
+		s.note(id, false)
+		log.Printf("watchers: user %s: %d of %d sources failed", id, res.Failed, res.Sources)
+		return
+	}
+	s.note(id, true)
+	if res.Sources > 0 || res.Created > 0 {
+		log.Printf("watchers: user %s: %d sources checked, %d signals created", id, res.Sources, res.Created)
+	}
+}
+
+func (s *Scheduler) note(id string, ok bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if ok {
+		delete(s.failures, id)
+	} else {
+		s.failures[id]++
+	}
+}
+
+// Browser alerts remain browser-only. Email watchers are checked independently
+// of an open tab, with source/external-ID deduplication and a durable per-source
+// UID checkpoint on every run. A slow pass delays the next one because the
+// ticker drops missed ticks.
+func RunLoop(ctx context.Context, d *db.DB) {
+	tick := time.NewTicker(5 * time.Minute)
+	defer tick.Stop()
+	s := NewScheduler()
+	s.RunOnce(ctx, d)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-tick.C:
+			s.RunOnce(ctx, d)
+		}
+	}
+}

internal/signals/signals.go

@@ -0,0 +1,443 @@
+package signals
+
+import (
+	"context"
+	"crypto/tls"
+	"database/sql"
+	"errors"
+	"fmt"
+	"github.com/andrew/go-computer/internal/config"
+	crypt "github.com/andrew/go-computer/internal/crypto"
+	"github.com/andrew/go-computer/internal/db"
+	"github.com/andrew/go-computer/internal/netguard"
+	"github.com/emersion/go-imap"
+	"github.com/emersion/go-imap/client"
+	"log"
+	"math"
+	"net"
+	"os"
+	"sort"
+	"strings"
+	"time"
+)
+
+type SignalInput struct {
+	ExternalID, Title, Summary string
+	OccurredAt                 time.Time
+}
+type Mailbox struct {
+	Host, Username, Password, Folder string
+	Port                             int
+	TLS                              bool
+}
+
+// AllowInsecure reports whether plaintext IMAP logins are permitted. Only
+// explicit development/test fixtures may set INTEGRATIONS_ALLOW_INSECURE_IMAP;
+// production configurations must always authenticate over verified TLS
+// (review NET-02). Never set it in production Compose.
+func AllowInsecure() bool { return os.Getenv("INTEGRATIONS_ALLOW_INSECURE_IMAP") == "1" }
+
+// checkTransport refuses password login on a plaintext socket (review NET-02).
+// Implicit TLS is the only production transport; there is no STARTTLS
+// downgrade path.
+func checkTransport(m Mailbox) error {
+	if !m.TLS && !AllowInsecure() {
+		return errors.New("signals: refusing plaintext IMAP connection; TLS is required")
+	}
+	return nil
+}
+
+// Connect dials through netguard, so user-supplied IMAP hosts cannot point
+// the server at internal infrastructure. The 15-second deadline covers
+// dial/TLS/login only and is cleared afterwards; later commands are bounded
+// by the caller's context (AfterFunc termination), not by a stale absolute
+// deadline.
+func Connect(ctx context.Context, m Mailbox) (*client.Client, error) {
+	if e := checkTransport(m); e != nil {
+		return nil, e
+	}
+	if m.Port == 0 {
+		m.Port = 993
+	}
+	// Keep outbound mailbox traffic on the normal IMAP ports. The fixture
+	// flags cover loopback fakes (and deliberate LAN deployments) that listen
+	// on an ephemeral port; production has no reason to dial, say, an SSH or
+	// database port through the netguard escape hatch.
+	if !netguard.AllowPrivate() && !AllowInsecure() && m.Port != 993 && m.Port != 143 {
+		return nil, errors.New("signals: mailbox port must be 143 or 993")
+	}
+	if m.Folder == "" {
+		m.Folder = "INBOX"
+	}
+	conn, e := netguard.DialContext(ctx, "tcp", net.JoinHostPort(m.Host, fmt.Sprint(m.Port)))
+	if e != nil {
+		return nil, e
+	}
+	if e = conn.SetDeadline(time.Now().Add(15 * time.Second)); e != nil {
+		conn.Close()
+		return nil, e
+	}
+	if m.TLS {
+		t := tls.Client(conn, &tls.Config{ServerName: m.Host, MinVersion: tls.VersionTLS12})
+		if e = t.HandshakeContext(ctx); e != nil {
+			conn.Close()
+			return nil, e
+		}
+		conn = t
+	}
+	c, e := client.New(conn)
+	if e != nil {
+		conn.Close()
+		return nil, e
+	}
+	if e = c.Login(m.Username, m.Password); e != nil {
+		c.Terminate()
+		return nil, e
+	}
+	if e = conn.SetDeadline(time.Time{}); e != nil {
+		c.Terminate()
+		return nil, e
+	}
+	return c, nil
+}
+func Search(ctx context.Context, m Mailbox, from, subject string) ([]SignalInput, error) {
+	c, e := Connect(ctx, m)
+	if e != nil {
+		return nil, e
+	}
+	stop := context.AfterFunc(ctx, func() { c.Terminate() })
+	defer stop()
+	defer c.Logout()
+	return SearchConn(c, m, from, subject)
+}
+
+// Checkpoint is the durable per-source ingestion state (review SYNC-02):
+// the folder UIDVALIDITY it was recorded against and the highest UID already
+// considered. A zero value means "nothing ingested yet".
+type Checkpoint struct{ UIDValidity, LastUID uint32 }
+
+func LoadCheckpoint(ctx context.Context, q db.Queryer, sourceID string) (Checkpoint, error) {
+	var cp Checkpoint
+	var validity, last int64
+	e := q.QueryRowContext(ctx, `SELECT "uidValidity","lastUid" FROM "SignalCheckpoint" WHERE "sourceId"=$1`, sourceID).Scan(&validity, &last)
+	if errors.Is(e, sql.ErrNoRows) {
+		return Checkpoint{}, nil
+	}
+	if e != nil {
+		return Checkpoint{}, e
+	}
+	cp.UIDValidity, cp.LastUID = uint32(validity), uint32(last)
+	return cp, nil
+}
+
+// SaveCheckpoint persists cp. Callers must only advance a checkpoint after
+// the batch's signals are durably handled. The update is monotonic (review
+// COR-4): when uidValidity is unchanged, lastUid only advances via GREATEST,
+// so a slower concurrent pass can never regress the checkpoint and force a
+// re-scan of messages already ingested.
+func SaveCheckpoint(ctx context.Context, q db.Queryer, sourceID string, cp Checkpoint) error {
+	_, e := q.ExecContext(ctx, `INSERT INTO "SignalCheckpoint"("sourceId","uidValidity","lastUid","updatedAt") VALUES($1,$2,$3,now()) ON CONFLICT("sourceId") DO UPDATE SET "uidValidity"=EXCLUDED."uidValidity","lastUid"=CASE WHEN "SignalCheckpoint"."uidValidity"=EXCLUDED."uidValidity" THEN GREATEST("SignalCheckpoint"."lastUid",EXCLUDED."lastUid") ELSE EXCLUDED."lastUid" END,"updatedAt"=now()`, sourceID, int64(cp.UIDValidity), int64(cp.LastUID))
+	return e
+}
+
+const (
+	// searchBatch bounds one UID FETCH round trip.
+	searchBatch = 50
+	// maxIngestBatches caps a single checkpointed pass at 2500 messages per
+	// source; a later pass continues from the persisted checkpoint.
+	maxIngestBatches = 50
+)
+
+// SearchConn runs one bounded search on an established connection and
+// reports the most recent matches. It is preview semantics for interactive
+// checks ("Check now", legacy mailbox watch): no checkpoint is consulted or
+// advanced. Durable ingestion uses SearchConnFrom. An empty criteria would
+// match the entire mailbox, so it yields nothing instead. On a mid-pass fetch
+// failure it returns the matches already fetched alongside the error.
+func SearchConn(c *client.Client, m Mailbox, from, subject string) ([]SignalInput, error) {
+	events, _, e := search(c, m, from, subject, nil)
+	return events, e
+}
+
+// SearchConnFrom processes only messages newer than cp and returns the
+// checkpoint to persist after — and only after — the returned inputs are
+// durably handled (review SYNC-02). A mid-pass fetch failure returns the
+// events fetched so far; callers must persist them WITHOUT saving the
+// checkpoint, so the failed batch is retried and deduplicated. Batches are processed until the source
+// is caught up or the per-pass cap is reached. When the folder's UIDVALIDITY
+// changed, the checkpoint resets and the mailbox is re-scanned; Message-ID
+// deduplication keeps re-scans from duplicating signals that carry one.
+func SearchConnFrom(c *client.Client, m Mailbox, from, subject string, cp Checkpoint) ([]SignalInput, Checkpoint, error) {
+	events, next, e := search(c, m, from, subject, &cp)
+	return events, next, e
+}
+
+func search(c *client.Client, m Mailbox, from, subject string, cp *Checkpoint) ([]SignalInput, Checkpoint, error) {
+	next := Checkpoint{}
+	if cp != nil {
+		next = *cp
+	}
+	if from == "" && subject == "" {
+		return []SignalInput{}, next, nil
+	}
+	folder := m.Folder
+	if folder == "" {
+		folder = "INBOX"
+	}
+	status, e := c.Select(folder, true)
+	if e != nil {
+		return nil, next, e
+	}
+	// Effective checkpoint (review CR-15): exactly one normalized checkpoint
+	// drives BOTH the UID filtering and the advancement below. When the
+	// folder's UIDVALIDITY changed, the previous UID space is meaningless —
+	// the effective checkpoint resets to zero so every UID in the new space
+	// is (re)ingested, and filtering must never consult the stale LastUID.
+	eff := Checkpoint{}
+	if cp != nil {
+		if cp.UIDValidity == status.UidValidity {
+			eff = *cp
+		}
+		next = eff
+	}
+	next.UIDValidity = status.UidValidity
+	criteria := imap.NewSearchCriteria()
+	if from != "" {
+		criteria.Header.Add("From", from)
+	}
+	if subject != "" {
+		criteria.Header.Add("Subject", subject)
+	}
+	// Review CR-19: constrain the server-side SEARCH to the UID window that
+	// can still hold unhandled messages, so a caught-up source does not
+	// re-transfer its whole match history every pass. "n:*" is inclusive of
+	// the mailbox's highest UID even when that UID is below n (RFC 3501
+	// seq-number semantics), so the defensive local filter below stays —
+	// the range is an optimization, never the correctness boundary.
+	if cp != nil && eff.LastUID > 0 && eff.LastUID < math.MaxUint32 {
+		criteria.Uid = new(imap.SeqSet)
+		criteria.Uid.AddRange(eff.LastUID+1, 0) // Stop 0 renders as "*"
+	}
+	uids, e := c.UidSearch(criteria)
+	if e != nil {
+		return nil, next, e
+	}
+	sort.Slice(uids, func(i, j int) bool { return uids[i] < uids[j] })
+	// truncated reports whether the per-pass cap shortened the UID list.
+	// Review COR-1: when it did, the SEARCH did NOT consider every matching
+	// message, so the high-water shortcut must stay off — otherwise UIDs past
+	// the cap would land below the new checkpoint and be dropped forever.
+	truncated := false
+	if cp != nil {
+		// Incremental ingestion: everything at or below the EFFECTIVE
+		// checkpoint has already been durably handled by an earlier pass
+		// (or the UID space was reset and nothing has — review CR-15).
+		filtered := uids[:0]
+		for _, uid := range uids {
+			if uid > eff.LastUID {
+				filtered = append(filtered, uid)
+			}
+		}
+		uids = filtered
+		if len(uids) > searchBatch*maxIngestBatches {
+			uids = uids[:searchBatch*maxIngestBatches]
+			truncated = true
+		}
+	} else if len(uids) > searchBatch {
+		// Preview mode: the most recent matches, best effort.
+		uids = uids[len(uids)-searchBatch:]
+	}
+	out := []SignalInput{}
+	for start := 0; start < len(uids); start += searchBatch {
+		batch := uids[start:min(start+searchBatch, len(uids))]
+		seq := new(imap.SeqSet)
+		seq.AddNum(batch...)
+		ch := make(chan *imap.Message, len(batch))
+		done := make(chan error, 1)
+		go func() { done <- c.UidFetch(seq, []imap.FetchItem{imap.FetchUid, imap.FetchEnvelope}, ch) }()
+		for msg := range ch {
+			if msg.Uid > next.LastUID {
+				next.LastUID = msg.Uid
+			}
+			if msg.Envelope == nil {
+				continue
+			}
+			env := msg.Envelope
+			id := env.MessageId
+			if id == "" {
+				id = fmt.Sprintf("%s:%d:%d", folder, status.UidValidity, msg.Uid)
+			}
+			names := []string{}
+			for _, a := range env.From {
+				names = append(names, a.Address())
+			}
+			out = append(out, SignalInput{ExternalID: id, Title: env.Subject, Summary: strings.Join(names, ", "), OccurredAt: env.Date})
+		}
+		if e := <-done; e != nil {
+			// Partial results (review follow-up): a fetch can fail after an
+			// earlier batch (or part of this one) already produced events.
+			// Return what was found; callers persist it but must NOT advance
+			// the checkpoint, so the failed batch is retried and deduped.
+			return out, next, e
+		}
+	}
+	if cp != nil && !truncated && status.UidNext > next.LastUID+1 {
+		// The SEARCH considered every message in the folder (the cap did not
+		// shorten it), so non-matching UIDs below the mailbox high-water
+		// mark can be skipped for good; anything arriving after this snapshot
+		// stays above the checkpoint. Review COR-1: this shortcut is OFF
+		// after truncation — remaining matches get a future pass instead of
+		// being skipped forever.
+		next.LastUID = status.UidNext - 1
+	}
+	return out, next, nil
+}
+
+func LoadMailbox(ctx context.Context, d *db.DB, user string) (Mailbox, error) {
+	v, e := db.One(ctx, d, `SELECT to_jsonb(m) FROM "ImapConfig" m WHERE "userId"=$1`, user)
+	if e != nil {
+		return Mailbox{}, e
+	}
+	s := func(k string) string { v, _ := v[k].(string); return v }
+	cfg := config.Load()
+	pw, e := crypt.Decrypt(s("password"), cfg.EncryptionSecret())
+	if e != nil {
+		// Review OPS-01: rows encrypted before ENCRYPTION_KEY existed used the
+		// auth-secret-derived key. Decrypt with the legacy key and re-encrypt
+		// under the dedicated key so the migration completes transparently.
+		legacy, legacyErr := crypt.Decrypt(s("password"), cfg.Secret())
+		if legacyErr != nil {
+			return Mailbox{}, e
+		}
+		pw = legacy
+		if cfg.EncryptionSecret() != cfg.Secret() {
+			if enc, encErr := crypt.Encrypt(pw, cfg.EncryptionSecret()); encErr == nil {
+				// Review CR-21: compare-and-swap against the EXACT ciphertext
+				// this pass decrypted. A password saved concurrently between
+				// the read and this write must win; zero updated rows means
+				// exactly that, not permission to overwrite.
+				res, updErr := d.ExecContext(ctx, `UPDATE "ImapConfig" SET password=$1 WHERE "userId"=$2 AND password=$3`, enc, user, s("password"))
+				if updErr != nil {
+					log.Printf("imap: re-encrypt stored password under ENCRYPTION_KEY: %v", updErr)
+				} else if n, _ := res.RowsAffected(); n == 0 {
+					log.Printf("imap: stored password changed during key migration; keeping the newer value")
+				}
+			}
+		}
+	}
+	port, _ := v["port"].(float64)
+	tls, _ := v["tls"].(bool)
+	return Mailbox{s("host"), s("username"), pw, s("folder"), int(port), tls}, nil
+}
+func PublicSourceTypes() []db.Object {
+	return []db.Object{{"type": "email", "label": "Email (IMAP)", "description": "Search your configured mailbox for matching messages", "stub": false, "configFields": []db.Object{{"key": "from", "label": "From", "placeholder": "sender@domain.com"}, {"key": "subject", "label": "Subject", "placeholder": "Subject contains"}}}}
+}
+
+// CheckResult is the structured outcome of one signal pass (review CR-16).
+// Sources counts the sources considered, Created the signals inserted, and
+// Failed the sources whose check did not complete successfully (mailbox
+// connection/search failure, signal insert failure, checkpoint failure or
+// status-write failure). Per-source detail is persisted on
+// SignalSource.lastStatus for the UI. The returned error is reserved for
+// pass-level infrastructure failures (loading the source list): per-source
+// failures live in Failed so scheduler backoff reacts to real integration
+// failures while a manual check still answers 200 with per-source outcomes.
+type CheckResult struct{ Sources, Created, Failed int }
+
+func RunSignalChecks(ctx context.Context, d *db.DB, user, bookmark string) (CheckResult, error) {
+	res := CheckResult{}
+	sources, e := db.Many(ctx, d, `SELECT to_jsonb(s) FROM "SignalSource" s WHERE "ownerId"=$1 AND enabled AND ($2='' OR "bookmarkId"=$2)`, user, bookmark)
+	if e != nil {
+		return res, e
+	}
+	res.Sources = len(sources)
+	if len(sources) == 0 {
+		return res, nil
+	}
+	// Load and connect once for the whole pass; every source of a user shares
+	// the same mailbox credentials.
+	m, connErr := LoadMailbox(ctx, d, user)
+	var c *client.Client
+	if connErr == nil {
+		c, connErr = Connect(ctx, m)
+		if connErr == nil {
+			stop := context.AfterFunc(ctx, func() { c.Terminate() })
+			defer stop()
+			defer c.Logout()
+		}
+	}
+	for _, src := range sources {
+		sourceID := fmt.Sprint(src["id"])
+		status := "Checked"
+		events := []SignalInput{}
+		e := connErr
+		var next Checkpoint
+		if e == nil {
+			cfg, _ := src["config"].(map[string]any)
+			from, _ := cfg["from"].(string)
+			subject, _ := cfg["subject"].(string)
+			// Incremental, checkpointed ingestion (review SYNC-02): only UIDs
+			// newer than the durable checkpoint are processed, in bounded
+			// batches, until the source is caught up. Review COR-3: a failed
+			// checkpoint load is surfaced as an error for this source instead
+			// of silently degrading to a zero checkpoint (full re-scan).
+			var cp Checkpoint
+			if cp, e = LoadCheckpoint(ctx, d, sourceID); e == nil {
+				events, next, e = SearchConnFrom(c, m, from, subject, cp)
+			}
+		}
+		// Persist whatever events this source produced. On a partial search
+		// failure the caller still inserts them; the checkpoint stays put so
+		// the failed batch is retried and ON CONFLICT dedupes the rest.
+		insertEvents := func() error {
+			for _, s := range events {
+				when := s.OccurredAt
+				if when.IsZero() {
+					when = time.Now()
+				}
+				r, err := d.ExecContext(ctx, `INSERT INTO "Signal"(id,"sourceType",title,summary,"externalId","occurredAt","sourceId","bookmarkId","ownerId") VALUES($1,'email',$2,$3,$4,$5,$6,$7,$8) ON CONFLICT("sourceId","externalId") DO NOTHING`, db.NewID(), s.Title, s.Summary, s.ExternalID, when, src["id"], src["bookmarkId"], user)
+				if err != nil {
+					return err
+				}
+				n, _ := r.RowsAffected()
+				res.Created += int(n)
+			}
+			return nil
+		}
+		if e != nil {
+			// Connection/search failures are recorded per source (lastStatus)
+			// and counted in Failed (review CR-16) — they are NOT folded into
+			// the pass-level error, so a manual check still answers 200 with
+			// the per-source outcome and db.ErrNotFound from a missing
+			// mailbox configuration cannot surface as a misleading 404. The
+			// scheduler reads Failed and backs off on repeated failures.
+			status = "Error: mailbox connection or search failed"
+			res.Failed++
+			// Review follow-up: a search that failed partway may still have
+			// found real messages. Keep them instead of discarding the whole
+			// pass; the checkpoint is deliberately not saved below.
+			if len(events) > 0 {
+				if err := insertEvents(); err != nil {
+					status = "Error: signal insert failed"
+				}
+			}
+		} else if err := insertEvents(); err != nil {
+			// Review COR-5: record the per-source failure in lastStatus
+			// and continue with the remaining sources instead of aborting
+			// the whole pass. The checkpoint does NOT advance, so the
+			// failed event and the rest of the batch are retried next
+			// pass (ON CONFLICT dedupes the ones already inserted).
+			status = "Error: signal insert failed"
+			res.Failed++
+		} else if err := SaveCheckpoint(ctx, d, sourceID, next); err != nil {
+			// Advance the checkpoint only after the batch is durably handled.
+			status = "Error: checkpoint save failed"
+			res.Failed++
+		}
+		// Record this source's status even on error so the UI can surface it.
+		if _, err := d.ExecContext(ctx, `UPDATE "SignalSource" SET "lastCheckedAt"=now(),"lastStatus"=$1,"updatedAt"=now() WHERE id=$2`, status, src["id"]); err != nil {
+			res.Failed++
+		}
+	}
+	return res, nil
+}

internal/signals/signals_test.go

@@ -0,0 +1,31 @@
+package signals
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+)
+
+// Review NET-02: mailbox credentials must never travel over a plaintext
+// socket unless the explicit development fixture flag is set.
+func TestPlaintextTransportRejected(t *testing.T) {
+	t.Setenv("INTEGRATIONS_ALLOW_INSECURE_IMAP", "")
+	if e := checkTransport(Mailbox{TLS: false}); e == nil {
+		t.Fatal("plaintext mailbox accepted without the development flag")
+	}
+	if e := checkTransport(Mailbox{TLS: true}); e != nil {
+		t.Fatalf("TLS mailbox rejected: %v", e)
+	}
+	// The guard runs before any dial, so Connect fails fast without network.
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	_, e := Connect(ctx, Mailbox{Host: "127.0.0.1", Port: 1, TLS: false})
+	if e == nil || !strings.Contains(e.Error(), "TLS is required") {
+		t.Fatalf("plaintext Connect was not refused up front: %v", e)
+	}
+	t.Setenv("INTEGRATIONS_ALLOW_INSECURE_IMAP", "1")
+	if e := checkTransport(Mailbox{TLS: false}); e != nil {
+		t.Fatalf("development fixture flag did not permit plaintext: %v", e)
+	}
+}