diff --git a/README.md b/README.md index 74fc3ae..dcea5ed 100644 --- a/README.md +++ b/README.md @@ -358,6 +358,14 @@ Yes, with caveats: **The Pipeline view** (`eraser pipeline` or the web UI) shows you exactly which brokers need manual attention. It's not fully automated, but it's free—and it does the tedious work of sending 700+ emails and tracking responses for you. +In the web Pipeline, select a status card to see matching requests or actions. +The **Responses** tab filters by response type and searches broker names, subjects, +and saved email bodies; **Action Required** also searches task notes. Filters stay +in the URL for refresh and browser back/forward. Clear "no records for you" replies +appear as **No data held**, including qualifying replies already saved before the +upgrade. + + --- ## How It Compares diff --git a/cmd/eraser/cmd_auto.go b/cmd/eraser/cmd_auto.go index 3a5cde1..d6a12f1 100644 --- a/cmd/eraser/cmd_auto.go +++ b/cmd/eraser/cmd_auto.go @@ -168,7 +168,7 @@ func sentSince(cfgPath string, start time.Time) int { if err != nil { return 0 } - store, err := history.NewStore(history.DBPathFor(cfgPath)) + store, err := openHistoryStore(history.DBPathFor(cfgPath)) if err != nil { return 0 } diff --git a/cmd/eraser/cmd_bounces.go b/cmd/eraser/cmd_bounces.go index 7d5bcba..060c35d 100644 --- a/cmd/eraser/cmd_bounces.go +++ b/cmd/eraser/cmd_bounces.go @@ -70,7 +70,7 @@ func runCleanupBounces(remove bool, days int) error { return fmt.Errorf("failed to load brokers: %w", err) } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } @@ -237,7 +237,7 @@ func runMarkBounced(brokerIDs []string, note string) error { return err } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } diff --git a/cmd/eraser/cmd_confirm.go b/cmd/eraser/cmd_confirm.go index 872d12f..183177b 100644 --- a/cmd/eraser/cmd_confirm.go +++ b/cmd/eraser/cmd_confirm.go @@ -74,7 +74,7 @@ func runConfirm(confirmURL, brokerID string, pending, validateDomain, dryRun boo return fmt.Errorf("failed to load brokers: %w", err) } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } diff --git a/cmd/eraser/cmd_export.go b/cmd/eraser/cmd_export.go index 50a18e2..9b71f02 100644 --- a/cmd/eraser/cmd_export.go +++ b/cmd/eraser/cmd_export.go @@ -76,7 +76,7 @@ func runExport(opts exportOptions) error { return err } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to open history: %w", err) } diff --git a/cmd/eraser/cmd_fill.go b/cmd/eraser/cmd_fill.go index d480ee6..5b8d3ab 100644 --- a/cmd/eraser/cmd_fill.go +++ b/cmd/eraser/cmd_fill.go @@ -128,7 +128,7 @@ func runFill(brokerID, formURL string, headless, headlessFlagSet, autoSubmit boo screenshotDir = filepath.Join(home, ".eraser", "screenshots") } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } diff --git a/cmd/eraser/cmd_mark_sent.go b/cmd/eraser/cmd_mark_sent.go index e24f9aa..91f87ae 100644 --- a/cmd/eraser/cmd_mark_sent.go +++ b/cmd/eraser/cmd_mark_sent.go @@ -83,7 +83,7 @@ func runMarkSent(args, regions, categories []string, dryRun bool) error { return nil } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to open history: %w", err) } diff --git a/cmd/eraser/cmd_monitor.go b/cmd/eraser/cmd_monitor.go index abf2e55..69b62f5 100644 --- a/cmd/eraser/cmd_monitor.go +++ b/cmd/eraser/cmd_monitor.go @@ -73,7 +73,7 @@ func runMonitor(days int, once bool, watch bool) error { return fmt.Errorf("failed to load brokers: %w", err) } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } @@ -151,6 +151,7 @@ func scanInbox(ctx context.Context, inboxCfg config.InboxConfig, brokerDB *broke fmt.Printf("📊 Summary for %s:\n", inboxCfg.Email) fmt.Printf(" Total responses: %d\n", summary.Total) fmt.Printf(" ✅ Success: %d\n", summary.Success) + fmt.Printf(" ℹ️ No data held: %d\n", summary.NoData) fmt.Printf(" 📝 Form required: %d\n", summary.FormRequired) fmt.Printf(" 🔗 Confirm required: %d\n", summary.ConfirmRequired) fmt.Printf(" ❌ Rejected: %d\n", summary.Rejected) @@ -182,7 +183,7 @@ func scanInbox(ctx context.Context, inboxCfg config.InboxConfig, brokerDB *broke func printClassifiedResponse(r inbox.ClassifiedResponse) { var icon string switch r.Type { - case inbox.ResponseSuccess: + case inbox.ResponseSuccess, inbox.ResponseNoData: icon = "✅" case inbox.ResponseFormRequired: icon = "📝" diff --git a/cmd/eraser/cmd_pipeline.go b/cmd/eraser/cmd_pipeline.go index a2a013d..ad15aab 100644 --- a/cmd/eraser/cmd_pipeline.go +++ b/cmd/eraser/cmd_pipeline.go @@ -32,7 +32,7 @@ func runPipelineStatus() error { return err } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to open history: %w", err) } @@ -59,6 +59,7 @@ func runPipelineStatus() error { fmt.Printf(" ✅ CAPTCHA solved: %d\n", pipelineStats[history.PipelineCaptchaSolved]) fmt.Printf(" 🔗 Awaiting confirmation: %d\n", pipelineStats[history.PipelineAwaitingConfirmation]) fmt.Printf(" ✅ Confirmed: %d\n", pipelineStats[history.PipelineConfirmed]) + fmt.Printf(" ℹ️ No data held: %d\n", pipelineStats[history.PipelineNoData]) fmt.Printf(" ❌ Rejected: %d\n", pipelineStats[history.PipelineRejected]) fmt.Printf(" 💥 Failed: %d\n", pipelineStats[history.PipelineFailed]) diff --git a/cmd/eraser/cmd_send.go b/cmd/eraser/cmd_send.go index c1ffc43..c80b059 100644 --- a/cmd/eraser/cmd_send.go +++ b/cmd/eraser/cmd_send.go @@ -157,7 +157,7 @@ func runSend() error { // Initialize history store early - needed for both the resend-cooldown // skip and the daily send cap below. - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } diff --git a/cmd/eraser/cmd_serve.go b/cmd/eraser/cmd_serve.go index 8a4951a..b1c5cd7 100644 --- a/cmd/eraser/cmd_serve.go +++ b/cmd/eraser/cmd_serve.go @@ -63,7 +63,7 @@ func runServe(port int) error { return fmt.Errorf("failed to load brokers: %w", err) } - store, err := history.NewStore(history.DBPathFor(configPath)) + store, err := openHistoryStore(history.DBPathFor(configPath)) if err != nil { return fmt.Errorf("failed to initialize history: %w", err) } diff --git a/cmd/eraser/cmd_status.go b/cmd/eraser/cmd_status.go index d99e72f..ed97ed7 100644 --- a/cmd/eraser/cmd_status.go +++ b/cmd/eraser/cmd_status.go @@ -36,7 +36,7 @@ func runStatus(limit int) error { return err } - store, err := history.NewStore(history.DBPathFor(resolveConfigPath())) + store, err := openHistoryStore(history.DBPathFor(resolveConfigPath())) if err != nil { return fmt.Errorf("failed to open history: %w", err) } diff --git a/cmd/eraser/history.go b/cmd/eraser/history.go new file mode 100644 index 0000000..0466acd --- /dev/null +++ b/cmd/eraser/history.go @@ -0,0 +1,22 @@ +package main + +import ( + "fmt" + + "github.com/drumandbytes/eraser/internal/history" + "github.com/drumandbytes/eraser/internal/inbox" +) + +// All production history openings share application-level data upgrades. The +// history package remains independent of the classifier that uses it. +func openHistoryStore(path string) (*history.Store, error) { + store, err := history.NewStore(path) + if err != nil { + return nil, err + } + if err := inbox.MigrateNoDataResponses(store); err != nil { + _ = store.Close() + return nil, fmt.Errorf("upgrade response classifications: %w", err) + } + return store, nil +} diff --git a/cmd/eraser/history_test.go b/cmd/eraser/history_test.go new file mode 100644 index 0000000..31ef8df --- /dev/null +++ b/cmd/eraser/history_test.go @@ -0,0 +1,31 @@ +package main + +import ( + "path/filepath" + "testing" + "time" + + "github.com/drumandbytes/eraser/internal/history" +) + +func TestOpenHistoryStoreUpgradesSavedNoDataReply(t *testing.T) { + path := filepath.Join(t.TempDir(), "history.db") + old, err := history.NewStore(path) + if err != nil { + t.Fatal(err) + } + r := history.BrokerResponse{BrokerID: "acme", ResponseType: "rejected", EmailSubject: "Reply", EmailBody: "We do not have any records of you.", ReceivedAt: time.Now()} + if err := old.AddBrokerResponse(&r); err != nil { + t.Fatal(err) + } + _ = old.Close() + upgraded, err := openHistoryStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = upgraded.Close() }() + got, err := upgraded.GetBrokerResponseByID(r.ID, "default") + if err != nil || got.ResponseType != "no_data" || got.EmailBody != r.EmailBody { + t.Fatalf("upgrade=%+v err=%v", got, err) + } +} diff --git a/docs/broker-replies.md b/docs/broker-replies.md index e116fb5..7da815b 100644 --- a/docs/broker-replies.md +++ b/docs/broker-replies.md @@ -26,7 +26,7 @@ Classifications match `ResponseType` in `internal/inbox/classifier.go`. | Reply says | Type | Action | |---|---|---| | Deletion done / suppressed / "we no longer hold your data" | `success` | none - archive the mail | -| "No record / no match for you" from a genuine broker | `rejected` | none - they're still a real broker, a future re-scrape may list you again | +| "No record / no match for you" from a genuine broker | `no_data` (No data held) | none - they're still a real broker, a future re-scrape may list you again | | "We are not a data broker" / "B2B only, not in scope" | `rejected` | remove the entry from `data/brokers.yaml`, then check the count floor (below) | | "We won't process this by email - use our web form / portal" | `form_required` | set `email: ""`, set/refresh `opt_out_url` to the form they name, add a dated `notes:` line | | Won't act without a government-ID document | - | set `category: requires-id`, add a dated `notes:` line (users exclude this category in bulk) | @@ -34,6 +34,14 @@ Classifications match `ResponseType` in `internal/inbox/classifier.go`. | Bounce / dead address | `bounced` | `eraser mark-bounced `, or set `email: ""` + dated note by hand. Never delete a real broker over a bounce - `MarkEmailUnreachable` in `internal/broker/broker.go` exists so the record survives for a later working address | | Refuses on a real statutory exemption (FCRA / GLBA / DPPA), genuine dead end | `rejected` | set `email: ""`, add a dated `notes:` line citing the basis; keep the row (see the `goodhire` entry) | +Clear no-record replies are separate from deletion confirmations and refusals. +Conditional replies or requests for identity verification still need review. +On upgrade, Eraser reclassifies qualifying saved replies once, using their stored +text without fetching mail. It only resolves the current request and superseded +pending work when the latest reply is confident; newer requests, newer tasks, +completed/skipped work, and manual pipeline progress are preserved. Original +messages and task notes remain in history. + "Remove the entry" vs. "blank the email" is a judgement call: delete when the company is genuinely not a broker and there's nothing to come back to; blank + note when they're a broker you just can't reach by email, or a dead end worth diff --git a/docs/code-patterns.md b/docs/code-patterns.md index 302ea16..d2ac084 100644 --- a/docs/code-patterns.md +++ b/docs/code-patterns.md @@ -5,8 +5,10 @@ - **Templates**: Go `text/template` with embedded files via `//go:embed`. - **CLI commands**: Each command has its own `cmd/eraser/cmd_*.go` file (e.g. `cmd_send.go`, `cmd_fill.go`) using Cobra; flags declared in the `*Cmd()` constructor, business logic in a separate `run*()` function so it's testable/callable independent of Cobra. `main.go` only holds root-command wiring and helpers shared by 3+ commands (config/profile/broker-path resolution). - **Regexes**: Compile once at package level (`var fooRe = regexp.MustCompile(...)`), never inside a function that runs per-email/per-loop-iteration. `internal/inbox/classifier.go` and `parser.go` follow this. -- **Reply classifier**: `internal/inbox/classifier.go` holds the upstream English/CCPA keyword patterns; `classifier_eu.go` appends GDPR-vocabulary and German/French patterns to the same package-level slices in `init()`, so both `ClassifyResponse` and `ClassifyBySubjectOnly` pick them up. Add new EU/multilingual patterns there, keeping non-English ones high-signal to avoid matching English text. A `rejected`/`success` verdict below 0.7 confidence is flagged `NeedsReview` (a wrong one silently drops a broker from the `eraser export` overdue list). +- **Reply classifier**: `internal/inbox/classifier.go` holds the upstream English/CCPA keyword patterns; `classifier_eu.go` appends GDPR-vocabulary and German/French patterns to the same package-level slices in `init()`, so both `ClassifyResponse` and `ClassifyBySubjectOnly` pick them up. Add new EU/multilingual patterns there, keeping non-English ones high-signal to avoid matching English text. A `rejected`/`success`/`no_data` verdict below 0.7 confidence is flagged `NeedsReview` (a wrong one silently drops a broker from the `eraser export` overdue list). - **SQLite**: `internal/history/history.go` has composite indexes (including `(profile_id, broker_id, sent_at DESC)`) for the "most recent record per broker, per profile" query pattern used throughout. `NewStore` opens the DB in WAL mode with `busy_timeout(5000)` so `eraser serve` and a CLI run can write at once; a copy/backup of `history.db` must include the `-wal` file (or be taken with the store closed), and WAL doesn't work on network filesystems. +- **History startup**: CLI commands use `openHistoryStore`, which opens the schema and runs the versioned, transactional saved-reply migration via `inbox.MigrateNoDataResponses`. Keep new entry points on this helper so upgrades do not require visiting the web UI. The migration is local-only and retryable after a rollback. +- **Pipeline UI**: counts and pending rows share `history.GetActionItems`; use it for Pipeline and Action Needed together. The Pipeline URL carries view, status, response type, search, review, and page filters. Response search filters the active profile in SQLite before pagination, including stored email bodies. - **Multi-profile scoping**: any new `history.Store` method that reads/writes per-user data should take a `profileID string` first parameter and filter by it (see [multi-profile.md](multi-profile.md)) - the two deliberate exceptions are `ClearBrokerResponses`/`GetAllBrokerResponses`, which operate on the one shared inbox across all profiles. ## Known quirks diff --git a/internal/evidence/evidence.go b/internal/evidence/evidence.go index fc47739..50f47bd 100644 --- a/internal/evidence/evidence.go +++ b/internal/evidence/evidence.go @@ -104,7 +104,7 @@ func legalBasisFor(templateName string) string { // actually engaging with the request (vs. silence, a bounce, or noise). func substantiveResponse(responseType string) bool { switch responseType { - case "success", "rejected", "form_required", "confirmation_required": + case "success", "rejected", "form_required", "confirmation_required", "no_data": return true default: return false @@ -230,7 +230,7 @@ func Build( ConfirmURL: r.ConfirmURL, NeedsReview: r.NeedsReview, }) - if substantiveResponse(r.ResponseType) { + if substantiveResponse(r.ResponseType) && (r.ResponseType != "no_data" || (!r.NeedsReview && r.Confidence >= 0.7)) { gotSubstantive = true } } diff --git a/internal/evidence/evidence_test.go b/internal/evidence/evidence_test.go index 400dc0a..aea6304 100644 --- a/internal/evidence/evidence_test.go +++ b/internal/evidence/evidence_test.go @@ -240,6 +240,18 @@ func TestJoinAddressSkipsBlankParts(t *testing.T) { } } +func TestNoDataEvidenceRequiresConfidentReply(t *testing.T) { + now := time.Now() + requests := []history.Record{{BrokerID: "acme", BrokerName: "Acme", Status: history.StatusSent, Template: "gdpr", SentAt: now.AddDate(0, -2, 0)}} + for _, review := range []bool{false, true} { + responses := []history.BrokerResponse{{BrokerID: "acme", ResponseType: "no_data", Confidence: 0.85, NeedsReview: review, ReceivedAt: now}} + report := Build(testProfile(), requests, responses, testBrokerDB(), testEngine(t), time.Time{}, now) + if got := findBroker(t, report, "acme"); got.PastDeadline != review || got.Responses[0].Type != "no_data" { + t.Fatalf("review=%v evidence=%+v", review, got) + } + } +} + func TestRenderHTML(t *testing.T) { now := time.Date(2026, 6, 1, 12, 0, 0, 0, time.UTC) p := testProfile() diff --git a/internal/history/history.go b/internal/history/history.go index 25f4743..7c3df4b 100644 --- a/internal/history/history.go +++ b/internal/history/history.go @@ -39,6 +39,7 @@ const ( PipelineCaptchaSolved PipelineStatus = "captcha_solved" PipelineAwaitingConfirmation PipelineStatus = "awaiting_confirmation" PipelineConfirmed PipelineStatus = "confirmed" + PipelineNoData PipelineStatus = "no_data" PipelineFailed PipelineStatus = "failed" PipelineRejected PipelineStatus = "rejected" ) @@ -75,7 +76,7 @@ type BrokerResponse struct { ProfileID string BrokerID string BrokerName string - ResponseType string // form_required, confirmation_required, success, rejected, pending, unknown + ResponseType string // form_required, confirmation_required, success, no_data, rejected, pending, bounced, unknown EmailFrom string EmailSubject string EmailBody string // Stored for reclassification @@ -464,6 +465,7 @@ func parseSQLiteTime(s string) (time.Time, error) { } formats := []string{ "2006-01-02 15:04:05.999999999 -0700 MST", + "2006-01-02 15:04:05.999999999 -0700 -0700", time.RFC3339Nano, "2006-01-02 15:04:05.999999999-07:00", "2006-01-02 15:04:05", @@ -479,16 +481,13 @@ func parseSQLiteTime(s string) (time.Time, error) { return time.Time{}, lastErr } -// parseFlexibleTimeString parses a nullable TEXT timestamp in RFC3339 or -// "2006-01-02 15:04:05". NULL gives the zero Time. +// parseFlexibleTimeString handles nullable timestamps from current and older +// SQLite/Go encodings, including numeric-only zones returned by IMAP. func parseFlexibleTimeString(s sql.NullString) time.Time { if !s.Valid { return time.Time{} } - if t, err := time.Parse(time.RFC3339, s.String); err == nil { - return t - } - t, _ := time.Parse("2006-01-02 15:04:05", s.String) + t, _ := parseSQLiteTime(s.String) return t } @@ -910,6 +909,7 @@ func (s *Store) GetResponseStats(profileID string) (map[string]int, error) { // FormWithStatus represents a form detected from email with its current fill status type FormWithStatus struct { + ResponseID int64 BrokerID string BrokerName string FormURL string @@ -923,30 +923,33 @@ type FormWithStatus struct { // GetFormsWithStatus returns all detected forms with their current status // for one profile. func (s *Store) GetFormsWithStatus(profileID string) ([]FormWithStatus, error) { + latest, err := s.LatestRequests(profileID) + if err != nil { + return nil, err + } + stages := make(map[string]PipelineStatus) + for _, r := range latest { + stages[r.BrokerID] = r.PipelineStatus + } profileID = normalizeProfileID(profileID) - // all three joins pinned to profile_id, so another profile's same broker never matches + // Pin the task join to the profile; another profile's same broker never matches. query := ` SELECT + br.id, br.broker_id, br.broker_name, br.form_url, br.email_subject, br.created_at as detected_at, COALESCE(pt.id, 0) as task_id, - COALESCE(pt.status, '') as task_status, - COALESCE(rr.pipeline_status, '') as pipeline_status + COALESCE(pt.status, '') as task_status FROM broker_responses br LEFT JOIN pending_tasks pt ON br.broker_id = pt.broker_id AND pt.profile_id = br.profile_id AND pt.task_type IN ('captcha', 'manual_form') - LEFT JOIN ( - SELECT broker_id, profile_id, pipeline_status - FROM removal_requests - WHERE id IN (SELECT MAX(id) FROM removal_requests WHERE profile_id = ? GROUP BY broker_id) - ) rr ON br.broker_id = rr.broker_id AND rr.profile_id = br.profile_id WHERE br.profile_id = ? AND br.form_url IS NOT NULL AND br.form_url != '' - ORDER BY br.created_at DESC + ORDER BY br.created_at DESC, br.id DESC, CASE WHEN pt.status = 'pending' THEN 0 ELSE 1 END, pt.id DESC ` - rows, err := s.db.Query(query, profileID, profileID) + rows, err := s.db.Query(query, profileID) if err != nil { return nil, fmt.Errorf("failed to query forms: %w", err) } @@ -957,10 +960,10 @@ func (s *Store) GetFormsWithStatus(profileID string) ([]FormWithStatus, error) { for rows.Next() { var f FormWithStatus - var taskStatus, pipelineStatus string + var taskStatus string - if err := rows.Scan(&f.BrokerID, &f.BrokerName, &f.FormURL, &f.EmailSubject, - &f.DetectedAt, &f.TaskID, &taskStatus, &pipelineStatus); err != nil { + if err := rows.Scan(&f.ResponseID, &f.BrokerID, &f.BrokerName, &f.FormURL, &f.EmailSubject, + &f.DetectedAt, &f.TaskID, &taskStatus); err != nil { return nil, fmt.Errorf("failed to scan form: %w", err) } @@ -969,9 +972,12 @@ func (s *Store) GetFormsWithStatus(profileID string) ([]FormWithStatus, error) { } seen[f.BrokerID] = true + pipelineStatus := string(stages[f.BrokerID]) f.PipelineStatus = PipelineStatus(pipelineStatus) - if taskStatus == "completed" { + if pipelineStatus == string(PipelineNoData) { + f.Status = "resolved" + } else if taskStatus == "completed" { f.Status = "filled" } else if taskStatus == "skipped" { f.Status = "skipped" @@ -1188,29 +1194,13 @@ func (s *Store) UpdatePipelineStatus(profileID, brokerID string, status Pipeline // GetPipelineStats returns counts by pipeline status for one profile func (s *Store) GetPipelineStats(profileID string) (map[PipelineStatus]int, error) { - profileID = normalizeProfileID(profileID) - query := `SELECT pipeline_status, COUNT(*) FROM removal_requests - WHERE profile_id = ? AND id IN (SELECT MAX(id) FROM removal_requests WHERE profile_id = ? GROUP BY broker_id) - GROUP BY pipeline_status` - - rows, err := s.db.Query(query, profileID, profileID) + latest, err := s.LatestRequests(profileID) if err != nil { - return nil, fmt.Errorf("failed to query pipeline stats: %w", err) + return nil, err } - defer func() { _ = rows.Close() }() - stats := make(map[PipelineStatus]int) - for rows.Next() { - var status sql.NullString - var count int - if err := rows.Scan(&status, &count); err != nil { - return nil, fmt.Errorf("failed to scan pipeline stat: %w", err) - } - if status.Valid { - stats[PipelineStatus(status.String)] = count - } else { - stats[PipelineEmailSent] = count - } + for _, r := range latest { + stats[r.PipelineStatus]++ } - return stats, rows.Err() + return stats, nil } diff --git a/internal/history/no_data.go b/internal/history/no_data.go new file mode 100644 index 0000000..5e8da39 --- /dev/null +++ b/internal/history/no_data.go @@ -0,0 +1,226 @@ +package history + +import ( + "database/sql" + "fmt" + "time" +) + +// NoDataClassifier keeps database migrations independent of the inbox package. +// It returns whether the saved reply says no data, its confidence, and whether +// that conclusion still needs a person's review. +type NoDataClassifier func(BrokerResponse) (bool, float64, bool) + +// MigrateNoData reclassifies saved replies once, atomically across profiles. +// Claiming the version inside the write transaction also serializes two app +// processes upgrading the same database. A failed migration is retryable. +func (s *Store) MigrateNoData(classify NoDataClassifier) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.Exec(`CREATE TABLE IF NOT EXISTS data_migrations (name TEXT PRIMARY KEY, applied_at DATETIME NOT NULL)`); err != nil { + return err + } + result, err := tx.Exec(`INSERT INTO data_migrations (name, applied_at) VALUES ('no_data_v1', ?) ON CONFLICT(name) DO NOTHING`, time.Now()) + if err != nil { + return err + } + n, err := result.RowsAffected() + if err != nil { + return err + } + if n == 0 { + return nil + } + + rows, err := tx.Query(`SELECT ` + brokerResponseCols + ` FROM broker_responses ORDER BY id`) + if err != nil { + return err + } + var replies []BrokerResponse + for rows.Next() { + r, err := scanBrokerResponse(rows) + if err != nil { + _ = rows.Close() + return err + } + replies = append(replies, *r) + } + err = rows.Err() + _ = rows.Close() + if err != nil { + return err + } + resolved := make(map[[2]string]bool) + for _, r := range replies { + matched, confidence, review := classify(r) + if !matched { + continue + } + if _, err := tx.Exec(`UPDATE broker_responses SET response_type = 'no_data', confidence = ?, needs_review = ? WHERE id = ? AND profile_id = ?`, confidence, review, r.ID, r.ProfileID); err != nil { + return err + } + if !review && confidence >= 0.7 { + resolved[[2]string{r.ProfileID, r.BrokerID}] = true + } + } + for key := range resolved { + if err := resolveNoData(tx, key[0], key[1], true); err != nil { + return err + } + } + return tx.Commit() +} + +// ResolveNoData reconciles only a confident, latest no-data reply. Call after +// inserting or reclassifying a reply; repeated calls are safe. +func (s *Store) ResolveNoData(profileID, brokerID string, reclassified bool) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + // Obtain the writer lock before reading, avoiding a WAL read/write upgrade. + if _, err := tx.Exec(`UPDATE broker_responses SET response_type = response_type WHERE profile_id = ? AND broker_id = ? AND response_type = 'no_data'`, normalizeProfileID(profileID), brokerID); err != nil { + return err + } + if err := resolveNoData(tx, normalizeProfileID(profileID), brokerID, reclassified); err != nil { + return err + } + return tx.Commit() +} + +func resolveNoData(tx *sql.Tx, profileID, brokerID string, upgrading bool) error { + replyRows, err := tx.Query(`SELECT `+brokerResponseCols+` FROM broker_responses WHERE profile_id = ? AND broker_id = ?`, profileID, brokerID) + if err != nil { + return err + } + var r *BrokerResponse + for replyRows.Next() { + candidate, err := scanBrokerResponse(replyRows) + if err != nil { + _ = replyRows.Close() + return err + } + if r == nil || candidate.ReceivedAt.After(r.ReceivedAt) || (candidate.ReceivedAt.Equal(r.ReceivedAt) && candidate.ID > r.ID) { + r = candidate + } + } + err = replyRows.Err() + _ = replyRows.Close() + if err != nil { + return err + } + if r == nil || r.ResponseType != "no_data" || r.NeedsReview || r.Confidence < 0.7 || r.ReceivedAt.IsZero() { + return nil + } + var requestID int64 + var sentAt sql.NullTime + var status sql.NullString + requestRows, err := tx.Query(`SELECT id, sent_at, pipeline_status FROM removal_requests WHERE profile_id = ? AND broker_id = ?`, profileID, brokerID) + if err != nil { + return err + } + for requestRows.Next() { + var id int64 + var sent sql.NullTime + var stage sql.NullString + if err := requestRows.Scan(&id, &sent, &stage); err != nil { + _ = requestRows.Close() + return err + } + if requestID == 0 || sent.Time.After(sentAt.Time) || (sent.Time.Equal(sentAt.Time) && id > requestID) { + requestID, sentAt, status = id, sent, stage + } + } + err = requestRows.Err() + _ = requestRows.Close() + if err != nil { + return err + } + if !sentAt.Valid || r.ReceivedAt.Before(sentAt.Time) { + return nil + } + + // There is no pipeline updated_at in legacy databases. On upgrade, only + // replace the old classifier's no-record outcomes; preserve manual progress. + switch PipelineStatus(status.String) { + case "", PipelineEmailSent, PipelineAwaitingResponse, PipelineRejected, PipelineNoData: + case PipelineFormRequired, PipelineAwaitingConfirmation, PipelineAwaitingCaptcha: + if upgrading { + return nil + } + default: + return nil + } + // SQLite cannot compare every legacy Go timestamp representation with + // julianday(). Read typed times and compare instants, including time zones. + rows, err := tx.Query(`SELECT id, status, created_at, completed_at FROM pending_tasks WHERE profile_id = ? AND broker_id = ?`, profileID, brokerID) + if err != nil { + return err + } + var superseded []int64 + later := false + for rows.Next() { + var id int64 + var taskStatus string + var created, completed sql.NullTime + if err := rows.Scan(&id, &taskStatus, &created, &completed); err != nil { + _ = rows.Close() + return err + } + if created.Time.After(r.ReceivedAt) || (completed.Valid && completed.Time.After(r.ReceivedAt)) { + later = true + } + if taskStatus == "pending" && created.Valid && !created.Time.Before(sentAt.Time) && !created.Time.After(r.ReceivedAt) { + superseded = append(superseded, id) + } + } + err = rows.Err() + _ = rows.Close() + if err != nil { + return err + } + if later { + return nil + } + if _, err := tx.Exec(`UPDATE removal_requests SET pipeline_status = 'no_data' WHERE id = ? AND profile_id = ?`, requestID, profileID); err != nil { + return err + } + // Keep the work and its notes as history; don't pretend a form was submitted. + for _, id := range superseded { + if _, err := tx.Exec(`UPDATE pending_tasks SET status = 'completed', completed_at = ?, notes = + CASE WHEN COALESCE(notes, '') = '' THEN 'Resolved: broker reports no data held.' ELSE notes || char(10) || 'Resolved: broker reports no data held.' END + WHERE id = ? AND profile_id = ? AND status = 'pending'`, r.ReceivedAt.UTC(), id, profileID); err != nil { + return err + } + } + rows, err = tx.Query(`SELECT `+brokerResponseCols+` FROM broker_responses WHERE profile_id = ? AND broker_id = ? AND needs_review = 1`, profileID, brokerID) + if err != nil { + return err + } + var reviews []int64 + for rows.Next() { + old, err := scanBrokerResponse(rows) + if err != nil { + _ = rows.Close() + return err + } + if !old.ReceivedAt.Before(sentAt.Time) && old.ReceivedAt.Before(r.ReceivedAt) { + reviews = append(reviews, old.ID) + } + } + err = rows.Err() + _ = rows.Close() + if err != nil { + return err + } + for _, id := range reviews { + if _, err := tx.Exec(`UPDATE broker_responses SET needs_review = 0 WHERE id = ? AND profile_id = ?`, id, profileID); err != nil { + return fmt.Errorf("resolve superseded response reviews: %w", err) + } + } + return nil +} diff --git a/internal/history/no_data_test.go b/internal/history/no_data_test.go new file mode 100644 index 0000000..3d884e2 --- /dev/null +++ b/internal/history/no_data_test.go @@ -0,0 +1,224 @@ +package history + +import ( + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestNoDataMigrationPreservesNewerWorkAndProfiles(t *testing.T) { + for _, tc := range []struct { + name string + stage PipelineStatus + newRequest, newReply, laterTask, uncertain bool + want PipelineStatus + }{ + {"old rejection", PipelineRejected, false, false, false, false, PipelineNoData}, + {"new request", PipelineRejected, true, false, false, false, PipelineEmailSent}, + {"new reply", PipelineAwaitingResponse, false, true, false, false, PipelineAwaitingResponse}, + {"manual progress", PipelineFormFilled, false, false, false, false, PipelineFormFilled}, + {"confirmed", PipelineConfirmed, false, false, false, false, PipelineConfirmed}, + {"new task", PipelineRejected, false, false, true, false, PipelineRejected}, + {"uncertain", PipelineRejected, false, false, false, true, PipelineRejected}, + } { + t.Run(tc.name, func(t *testing.T) { + s := newTestStore(t) + sent := time.Now().Add(-48 * time.Hour) + received := sent.Add(time.Hour) + for _, p := range []string{"a", "b"} { + if err := s.Add(&Record{ProfileID: p, BrokerID: "acme", BrokerName: "Acme", Status: StatusSent, SentAt: sent}); err != nil { + t.Fatal(err) + } + if err := s.UpdatePipelineStatus(p, "acme", tc.stage); err != nil { + t.Fatal(err) + } + } + r := BrokerResponse{ProfileID: "a", BrokerID: "acme", BrokerName: "Acme", ResponseType: "rejected", EmailSubject: "Reply", EmailBody: "Original no-records body", ReceivedAt: received} + if err := s.AddBrokerResponse(&r); err != nil { + t.Fatal(err) + } + task := PendingTask{ProfileID: "a", BrokerID: "acme", BrokerName: "Acme", TaskType: TaskReview, Notes: "Original note"} + if err := s.AddPendingTask(&task); err != nil { + t.Fatal(err) + } + created := sent.Add(30 * time.Minute) + if tc.laterTask { + created = received.Add(time.Hour) + } + if _, err := s.db.Exec(`UPDATE pending_tasks SET created_at = ? WHERE id = ?`, created, task.ID); err != nil { + t.Fatal(err) + } + if tc.newRequest { + if err := s.Add(&Record{ProfileID: "a", BrokerID: "acme", BrokerName: "Acme", Status: StatusSent, SentAt: received.Add(time.Hour)}); err != nil { + t.Fatal(err) + } + } + if tc.newReply { + if err := s.AddBrokerResponse(&BrokerResponse{ProfileID: "a", BrokerID: "acme", ResponseType: "pending", ReceivedAt: received.Add(time.Hour)}); err != nil { + t.Fatal(err) + } + } + classify := func(r BrokerResponse) (bool, float64, bool) { return r.EmailBody != "", 0.85, tc.uncertain } + if err := s.MigrateNoData(classify); err != nil { + t.Fatal(err) + } + got, _ := s.GetBrokerResponseByID(r.ID, "a") + if got.ResponseType != "no_data" || got.EmailBody != r.EmailBody || got.NeedsReview != tc.uncertain { + t.Fatalf("reply=%+v", got) + } + a, _ := s.LatestRequests("a") + b, _ := s.LatestRequests("b") + if a[0].PipelineStatus != tc.want || b[0].PipelineStatus != tc.stage { + t.Fatalf("a=%s want=%s; b=%s want=%s", a[0].PipelineStatus, tc.want, b[0].PipelineStatus, tc.stage) + } + gotTask, _ := s.GetPendingTaskByID(task.ID, "a") + if tc.want == PipelineNoData { + if gotTask.Status != "completed" || !strings.Contains(gotTask.Notes, "Original note") || !strings.Contains(gotTask.Notes, "no data held") { + t.Fatalf("resolved task=%+v", gotTask) + } + } else if gotTask.Status != "pending" { + t.Fatal("changed work that was not superseded") + } + if err := s.MigrateNoData(func(BrokerResponse) (bool, float64, bool) { t.Fatal("migration ran twice"); return false, 0, false }); err != nil { + t.Fatal(err) + } + }) + } +} + +func TestNoDataMigrationRollsBackAndRetries(t *testing.T) { + s := newTestStore(t) + r := BrokerResponse{ProfileID: "a", BrokerID: "acme", ResponseType: "rejected", EmailBody: "Raw body", ReceivedAt: time.Now()} + if err := s.AddBrokerResponse(&r); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`CREATE TRIGGER stop_upgrade BEFORE UPDATE OF response_type ON broker_responses BEGIN SELECT RAISE(ABORT, 'test rollback'); END`); err != nil { + t.Fatal(err) + } + classify := func(BrokerResponse) (bool, float64, bool) { return true, 0.85, false } + if err := s.MigrateNoData(classify); err == nil { + t.Fatal("expected migration failure") + } + got, _ := s.GetBrokerResponseByID(r.ID, "a") + if got.ResponseType != "rejected" { + t.Fatal("partial migration committed") + } + if _, err := s.db.Exec(`DROP TRIGGER stop_upgrade`); err != nil { + t.Fatal(err) + } + if err := s.MigrateNoData(classify); err != nil { + t.Fatal(err) + } + got, _ = s.GetBrokerResponseByID(r.ID, "a") + if got.ResponseType != "no_data" { + t.Fatal("failed migration was not retried") + } +} + +func TestNoDataMigrationConcurrentStores(t *testing.T) { + path := filepath.Join(t.TempDir(), "history.db") + a, err := NewStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = a.Close() }() + b, err := NewStore(path) + if err != nil { + t.Fatal(err) + } + defer func() { _ = b.Close() }() + if err := a.AddBrokerResponse(&BrokerResponse{BrokerID: "acme", ResponseType: "rejected", ReceivedAt: time.Now()}); err != nil { + t.Fatal(err) + } + var calls atomic.Int32 + classify := func(BrokerResponse) (bool, float64, bool) { calls.Add(1); return true, 0.85, false } + start := make(chan struct{}) + results := make(chan error, 2) + for _, s := range []*Store{a, b} { + go func() { <-start; results <- s.MigrateNoData(classify) }() + } + close(start) + for range 2 { + if err := <-results; err != nil { + t.Fatal(err) + } + } + if calls.Load() != 1 { + t.Fatalf("classifier ran %d times", calls.Load()) + } +} + +func TestNoDataResolutionUsesInstantsAndPreservesTaskHistory(t *testing.T) { + s := newTestStore(t) + base := time.Date(2026, 1, 1, 10, 0, 0, 0, time.UTC) + west := time.FixedZone("EST", -5*60*60) + // The latest send and reply sort earlier as strings, and were imported + // before their older counterparts. Neither insertion order nor local time + // is sufficient to select the current request/reply. + current := Record{BrokerID: "acme", Status: StatusSent, SentAt: base.Add(2 * time.Hour).In(west)} + older := Record{BrokerID: "acme", Status: StatusSent, SentAt: base} + for _, r := range []*Record{¤t, &older} { + if err := s.Add(r); err != nil { + t.Fatal(err) + } + } + received := base.Add(3 * time.Hour).In(time.FixedZone("", -5*60*60)) + for _, r := range []BrokerResponse{ + {BrokerID: "acme", ResponseType: "no_data", Confidence: 0.85, ReceivedAt: received}, + {BrokerID: "acme", ResponseType: "pending", NeedsReview: true, ReceivedAt: base.Add(150 * time.Minute)}, + } { + if err := s.AddBrokerResponse(&r); err != nil { + t.Fatal(err) + } + } + var tasks []PendingTask + for i, status := range []string{"pending", "completed", "skipped"} { + task := PendingTask{BrokerID: "acme", TaskType: TaskReview, Notes: "Original note"} + if err := s.AddPendingTask(&task); err != nil { + t.Fatal(err) + } + created := base.Add(time.Duration(121+i) * time.Minute) + var completed any + if status != "pending" { + completed = base.Add(140 * time.Minute) + } + if _, err := s.db.Exec(`UPDATE pending_tasks SET status=?,created_at=?,completed_at=? WHERE id=?`, status, created, completed, task.ID); err != nil { + t.Fatal(err) + } + task.Status = status + tasks = append(tasks, task) + } + if err := s.ResolveNoData("default", "acme", true); err != nil { + t.Fatal(err) + } + latest, err := s.LatestRequests("default") + if err != nil || len(latest) != 1 || latest[0].ID != current.ID || latest[0].PipelineStatus != PipelineNoData { + t.Fatalf("latest=%+v err=%v", latest, err) + } + var oldStage string + if err := s.db.QueryRow(`SELECT pipeline_status FROM removal_requests WHERE id=?`, older.ID).Scan(&oldStage); err != nil { + t.Fatal(err) + } + if oldStage != string(PipelineEmailSent) { + t.Fatalf("changed previous request: %s", oldStage) + } + for _, task := range tasks { + got, err := s.GetPendingTaskByID(task.ID, "default") + if err != nil { + t.Fatal(err) + } + if task.Status == "pending" { + if got.Status != "completed" || !strings.Contains(got.Notes, "no data held") { + t.Fatalf("pending work not resolved: %+v", got) + } + } else if got.Status != task.Status || got.Notes != task.Notes || !got.CompletedAt.Valid || !got.CompletedAt.Time.Equal(base.Add(140*time.Minute)) { + t.Fatalf("changed task history: %+v", got) + } + } + reviews, err := s.GetBrokerResponses("default", "", true, -1) + if err != nil || len(reviews) != 0 { + t.Fatalf("superseded reviews=%+v err=%v", reviews, err) + } +} diff --git a/internal/history/pipeline.go b/internal/history/pipeline.go new file mode 100644 index 0000000..c3d47e1 --- /dev/null +++ b/internal/history/pipeline.go @@ -0,0 +1,150 @@ +package history + +import ( + "fmt" + "sort" + "strings" +) + +// ActionItems is the common source for action rows and counters in both web +// pages. Forms represented by tasks appear only once. Reviews are independent +// actions and are not restricted to the response type "unknown". +type ActionItems struct { + Forms []FormWithStatus + Tasks []PendingTask + Reviews []BrokerResponse +} + +func (a ActionItems) Total() int { return len(a.Forms) + len(a.Tasks) + len(a.Reviews) } +func (a ActionItems) NeedsReview() int { + n := len(a.Reviews) + for _, t := range a.Tasks { + if t.TaskType == TaskReview { + n++ + } + } + return n +} +func (a ActionItems) FormsToFill() int { + n := len(a.Forms) + for _, t := range a.Tasks { + if t.TaskType == TaskManualForm { + n++ + } + } + return n +} + +func (s *Store) GetActionItems(profileID string) (ActionItems, error) { + var a ActionItems + forms, err := s.GetFormsWithStatus(profileID) + if err != nil { + return a, err + } + a.Tasks, err = s.GetPendingTasks(profileID, "", "pending") + if err != nil { + return a, err + } + withTask := make(map[string]bool) + for _, t := range a.Tasks { + if t.TaskType == TaskManualForm || t.TaskType == TaskCaptcha { + withTask[t.BrokerID] = true + } + } + for _, f := range forms { + if f.Status == "pending" && !withTask[f.BrokerID] { + a.Forms = append(a.Forms, f) + } + } + a.Reviews, err = s.GetBrokerResponses(profileID, "", true, -1) + return a, err +} + +// ResponseQuery searches the entire active profile before applying pagination. +type ResponseQuery struct { + Type string + NeedsReview bool + Search string + Limit int + Offset int +} + +func responseWhere(profileID string, q ResponseQuery) (string, []any) { + where := "profile_id = ?" + args := []any{normalizeProfileID(profileID)} + if q.Type != "" { + where += " AND response_type = ?" + args = append(args, q.Type) + } + if q.NeedsReview { + where += " AND needs_review = 1" + } + if search := strings.TrimSpace(q.Search); search != "" { + // Match Unicode case-insensitively, with %, _ and backslashes treated literally. + where += ` AND (eraser_contains(broker_name, ?) OR eraser_contains(broker_id, ?) OR + eraser_contains(email_from, ?) OR eraser_contains(email_subject, ?) OR eraser_contains(email_body, ?))` + for range 5 { + args = append(args, search) + } + } + return where, args +} + +func (s *Store) QueryBrokerResponses(profileID string, q ResponseQuery) ([]BrokerResponse, int, error) { + where, args := responseWhere(profileID, q) + var total int + if err := s.db.QueryRow(`SELECT COUNT(*) FROM broker_responses WHERE `+where, args...).Scan(&total); err != nil { + return nil, 0, err + } + rows, err := s.queryBrokerResponses(`SELECT `+brokerResponseCols+` FROM broker_responses WHERE `+where+` ORDER BY received_at DESC, id DESC LIMIT ? OFFSET ?`, append(args, q.Limit, q.Offset)...) + return rows, total, err +} + +// MatchingResponseIDs lets action searches include a form/review's underlying +// email text without loading every body into the web response. +func (s *Store) MatchingResponseIDs(profileID, search string) (map[int64]bool, error) { + where, args := responseWhere(profileID, ResponseQuery{Search: search}) + rows, err := s.db.Query(`SELECT id FROM broker_responses WHERE `+where, args...) + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + ids := make(map[int64]bool) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, err + } + ids[id] = true + } + return ids, rows.Err() +} + +// LatestRequests uses the same most-recent-send ordering as status updates. +// Historical sends do not inflate stage counters or their drill-down lists. +func (s *Store) LatestRequests(profileID string) ([]Record, error) { + all, err := s.GetAllRequests(profileID) + if err != nil { + return nil, fmt.Errorf("latest pipeline requests: %w", err) + } + sort.Slice(all, func(i, j int) bool { + if all[i].SentAt.Equal(all[j].SentAt) { + return all[i].ID < all[j].ID + } + return all[i].SentAt.Before(all[j].SentAt) + }) + seen := make(map[string]bool) + var latest []Record + for i := len(all) - 1; i >= 0; i-- { + r := all[i] + if seen[r.BrokerID] { + continue + } + seen[r.BrokerID] = true + if r.PipelineStatus == "" { + r.PipelineStatus = PipelineEmailSent + } + latest = append(latest, r) + } + return latest, nil +} diff --git a/internal/history/queries_test.go b/internal/history/queries_test.go index df9df99..287a301 100644 --- a/internal/history/queries_test.go +++ b/internal/history/queries_test.go @@ -175,6 +175,7 @@ func TestParseTimeHelpers(t *testing.T) { want := time.Date(2026, 3, 4, 5, 6, 7, 0, time.UTC) for _, in := range []string{ "2026-03-04 05:06:07 +0000 UTC m=+0.000000001", + "2026-03-04 00:06:07 -0500 -0500", "2026-03-04T05:06:07Z", "2026-03-04 05:06:07+00:00", "2026-03-04 05:06:07", @@ -191,7 +192,7 @@ func TestParseTimeHelpers(t *testing.T) { if got := parseFlexibleTimeString(sql.NullString{}); !got.IsZero() { t.Errorf("NULL = %v, want zero", got) } - for _, in := range []string{"2026-03-04T05:06:07Z", "2026-03-04 05:06:07"} { + for _, in := range []string{"2026-03-04T05:06:07Z", "2026-03-04 05:06:07", "2026-03-04 00:06:07 -0500 -0500"} { if got := parseFlexibleTimeString(sql.NullString{String: in, Valid: true}); !got.Equal(want) { t.Errorf("parseFlexibleTimeString(%q) = %v", in, got) } diff --git a/internal/history/search.go b/internal/history/search.go new file mode 100644 index 0000000..7d0b865 --- /dev/null +++ b/internal/history/search.go @@ -0,0 +1,21 @@ +package history + +import ( + "database/sql/driver" + "strings" + + "modernc.org/sqlite" +) + +func init() { + // SQLite's built-in lower() only handles ASCII. Broker names and replies + // also use accented letters; keep response and Go-side task search aligned. + sqlite.MustRegisterDeterministicScalarFunction("eraser_contains", 2, func(_ *sqlite.FunctionContext, args []driver.Value) (driver.Value, error) { + value, _ := args[0].(string) + query, _ := args[1].(string) + if strings.Contains(strings.ToLower(value), strings.ToLower(query)) { + return int64(1), nil + } + return int64(0), nil + }) +} diff --git a/internal/inbox/classifier.go b/internal/inbox/classifier.go index f9bda46..7330c57 100644 --- a/internal/inbox/classifier.go +++ b/internal/inbox/classifier.go @@ -12,6 +12,7 @@ const ( ResponseFormRequired ResponseType = "form_required" // Need to fill out a form ResponseConfirmationRequired ResponseType = "confirmation_required" // Need to click confirmation link ResponseSuccess ResponseType = "success" // Removal confirmed/completed + ResponseNoData ResponseType = "no_data" // Broker holds no matching data ResponseRejected ResponseType = "rejected" // Request denied ResponsePending ResponseType = "pending" // Processing, will follow up ResponseBounced ResponseType = "bounced" // Email bounced - invalid address @@ -88,27 +89,16 @@ var ( rejectionPatterns = []*regexp.Regexp{ regexp.MustCompile(`(?i)(cannot|can't|unable\s+to)\s+(process|complete|fulfill)\s+(your\s+)?request`), regexp.MustCompile(`(?i)request\s+(has\s+been\s+)?(denied|rejected|declined)`), - regexp.MustCompile(`(?i)do\s+not\s+have\s+(any\s+)?(data|information|records)\s+(about|for|on)\s+you`), - regexp.MustCompile(`(?i)not\s+found\s+in\s+our\s+(system|database|records)`), - regexp.MustCompile(`(?i)no\s+(matching\s+)?(records?|data|information)\s+found`), regexp.MustCompile(`(?i)exempt\s+from\s+(CCPA|GDPR|this\s+request)`), - // No data found patterns (from real broker emails) - regexp.MustCompile(`(?i)(do\s+not|don't)\s+have\s+(any\s+)?record`), - regexp.MustCompile(`(?i)no\s+(matching\s+)?record\s+(of\s+)?(a\s+)?report`), - regexp.MustCompile(`(?i)maintains?\s+no\s+(files?|records?|data)`), // Service discontinued patterns regexp.MustCompile(`(?i)no\s+longer\s+(registered|operating)\s+(as\s+)?(an?\s+)?(active\s+)?data\s+broker`), regexp.MustCompile(`(?i)(this\s+)?(email|inbox)(\s+\w+)?\s+(is\s+)?(no\s+longer|being\s+retired)`), regexp.MustCompile(`(?i)service\s+offerings?\s+no\s+longer\s+include`), // Additional rejection patterns from real emails - regexp.MustCompile(`(?i)(we\s+)?(have\s+)?no\s+data\s+(linked|associated|related)\s+to\s+(your|this)`), - regexp.MustCompile(`(?i)not\s+identified\s+in\s+our\s+database`), regexp.MustCompile(`(?i)(we\s+are|we're)\s+a\s+b2b\s+(platform|company|business)`), - regexp.MustCompile(`(?i)has\s+never\s+existed\s+in\s+our\s+database`), regexp.MustCompile(`(?i)consumer\s+reporting\s+agenc(y|ies)\s+(is|are)\s+exempt`), regexp.MustCompile(`(?i)fair\s+credit\s+reporting\s+act.{0,30}exempt`), regexp.MustCompile(`(?i)we\s+do\s+not\s+(remove|delete)\s+data\s+by\s+request`), - regexp.MustCompile(`(?i)(your\s+)?(email|name|address|information)\s+was\s+not\s+identified`), // Wrong email address patterns regexp.MustCompile(`(?i)(this\s+)?(email|inbox)\s+(address\s+)?(is\s+)?not\s+(a\s+)?(mechanism|intended)\s+(for|to)`), regexp.MustCompile(`(?i)not\s+intended\s+for\s+(the\s+)?(submission|handling)\s+of\s+privacy`), @@ -175,9 +165,7 @@ var ( // Subject-specific rejection patterns subjectRejectionPatterns = []*regexp.Regexp{ - regexp.MustCompile(`(?i)not\s+found`), - regexp.MustCompile(`(?i)no\s+record`), - regexp.MustCompile(`(?i)unable\s+to\s+(locate|find|process)`), + regexp.MustCompile(`(?i)unable\s+to\s+process`), regexp.MustCompile(`(?i)request\s+(denied|rejected)`), } @@ -268,6 +256,19 @@ func ClassifyResponse(email *Email) ClassifiedResponse { return result } + if matched, review := noDataVerdict(subject, emailBody(email)); matched { + result.Type = ResponseNoData + result.Confidence = 0.85 + result.NeedsReview = review + if review { + result.Confidence = 0.5 + result.FormURL = GetPrimaryFormURL(result.URLs, email.FromDomain) + result.ConfirmURL = GetPrimaryConfirmationURL(result.URLs, email.FromDomain) + } + result.Reason = getClassificationReason(result.Type, 0) + return result + } + // Score each category scores := map[ResponseType]int{ ResponseSuccess: 0, @@ -412,6 +413,8 @@ func getClassificationReason(responseType ResponseType, score int) string { switch responseType { case ResponseSuccess: return "Email indicates removal request was completed successfully" + case ResponseNoData: + return "Broker reports holding no matching personal data" case ResponseFormRequired: return "Email contains link to opt-out form that needs to be filled" case ResponseConfirmationRequired: @@ -461,6 +464,12 @@ func stripHTML(html string) string { // ClassifyBySubjectOnly classifies based on subject line only (for reclassifying database records) // Returns the response type, confidence, and whether it needs review func ClassifyBySubjectOnly(subject string) (ResponseType, float64, bool) { + if matched, review := noDataVerdict(subject, ""); matched { + if review { + return ResponseNoData, 0.5, true + } + return ResponseNoData, 0.7, false + } subject = strings.ToLower(subject) // Score each category based on subject patterns @@ -546,6 +555,7 @@ type Summary struct { FormRequired int ConfirmRequired int Rejected int + NoData int Pending int Bounced int Unknown int @@ -566,6 +576,8 @@ func SummarizeResponses(responses []ClassifiedResponse) Summary { summary.ConfirmRequired++ case ResponseRejected: summary.Rejected++ + case ResponseNoData: + summary.NoData++ case ResponsePending: summary.Pending++ case ResponseBounced: diff --git a/internal/inbox/classifier_eu.go b/internal/inbox/classifier_eu.go index 07742a2..b020303 100644 --- a/internal/inbox/classifier_eu.go +++ b/internal/inbox/classifier_eu.go @@ -48,12 +48,11 @@ func init() { regexp.MustCompile(`(?i)copie\s+de\s+votre\s+(pièce\s+d['’]identité|carte\s+d['’]identité|passeport)`), ) - rejectionPatterns = append(rejectionPatterns, + noDataPatterns = append(noDataPatterns, regexp.MustCompile(`(?i)(do\s+not|don't|does\s+not)\s+(process|hold|store)\s+any\s+personal\s+data\s+(relating|related|concerning|pertaining)\s+to\s+you`), regexp.MustCompile(`(?i)no\s+personal\s+data\s+(concerning|relating\s+to|about|pertaining\s+to)\s+you`), regexp.MustCompile(`(?i)you\s+are\s+not\s+(present\s+|listed\s+)?in\s+our\s+(database|records|systems)`), regexp.MustCompile(`(?i)unable\s+to\s+identify\s+you\s+in\s+our\s+(records|database|systems)`), - regexp.MustCompile(`(?i)manifestly\s+(unfounded|excessive)`), // German regexp.MustCompile(`(?i)(wir\s+)?(verarbeiten|speichern|haben)\s+keine\s+(personenbezogenen\s+)?daten\s+(zu|über|von)\s+ihnen`), regexp.MustCompile(`(?i)sie\s+sind\s+nicht\s+in\s+unserer\s+datenbank`), @@ -89,9 +88,9 @@ func init() { regexp.MustCompile(`(?i)(supprimé|effacé|effacement\s+effectué)`), ) - subjectRejectionPatterns = append(subjectRejectionPatterns, + rejectionPatterns = append(rejectionPatterns, regexp.MustCompile(`(?i)manifestly\s+(unfounded|excessive)`)) + subjectNoDataPatterns = append(subjectNoDataPatterns, regexp.MustCompile(`(?i)keine\s+daten`), regexp.MustCompile(`(?i)aucune\s+donnée`), - regexp.MustCompile(`(?i)(nicht\s+gefunden|introuvable)`), ) } diff --git a/internal/inbox/classifier_eu_test.go b/internal/inbox/classifier_eu_test.go index 0fce684..85f03a8 100644 --- a/internal/inbox/classifier_eu_test.go +++ b/internal/inbox/classifier_eu_test.go @@ -23,7 +23,7 @@ func TestGDPRAndMultilingualClassification(t *testing.T) { { name: "no personal data concerning you", body: "Following our checks, we hold no personal data concerning you and there is nothing to erase.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "manifestly unfounded", @@ -56,7 +56,7 @@ func TestGDPRAndMultilingualClassification(t *testing.T) { { name: "German - no data", body: "Wir verarbeiten keine personenbezogenen Daten zu Ihnen. Sie sind nicht in unserer Datenbank.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "German - acknowledgement", @@ -83,7 +83,7 @@ func TestGDPRAndMultilingualClassification(t *testing.T) { { name: "French - no data", body: "Nous ne détenons aucune donnée vous concernant dans nos fichiers.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "French - acknowledgement", @@ -125,7 +125,7 @@ func TestEUPatternsDoNotRegressEnglish(t *testing.T) { expected ResponseType }{ {"please submit your request at https://example.com/dsar", ResponseFormRequired}, - {"we do not have any record of a report in our system", ResponseRejected}, + {"we do not have any record of a report in our system", ResponseNoData}, {"we have received your request. One of our Privacy Specialists will reach out", ResponsePending}, {"Please click here to confirm your request", ResponseConfirmationRequired}, {"your data has been removed and we no longer hold your information", ResponseSuccess}, diff --git a/internal/inbox/classifier_no_data.go b/internal/inbox/classifier_no_data.go new file mode 100644 index 0000000..8b40342 --- /dev/null +++ b/internal/inbox/classifier_no_data.go @@ -0,0 +1,86 @@ +package inbox + +import ( + "regexp" + "strings" +) + +func emailBody(email *Email) string { + if email.Body != "" { + return email.Body + } + return email.HTMLBody +} + +// A broker reporting no records is different from refusing an erasure request. +// These patterns are also used when upgrading saved replies. +var noDataPatterns = []*regexp.Regexp{ + regexp.MustCompile(`(?i)\b(have|hold|store)\s+no\s+(matching\s+)?(personal\s+)?(data|information|records?)\s+(about|for|on|of|concerning|relating\s+to)\s+(you|your)\b`), + regexp.MustCompile(`(?i)(we\s+)?(do\s+not|don't)\s+(have|hold|store)\s+(any\s+)?(personal\s+)?(data|information|records?)\s+(about|for|on|of|relating\s+to)\s+you`), + regexp.MustCompile(`(?i)no\s+(matching\s+)?(records?|data|information)\s+(was\s+|were\s+)?found`), + regexp.MustCompile(`(?i)(you|your\s+(data|information|email|name|address))\s+(was\s+|were\s+|is\s+|are\s+)?not\s+(found|identified)\s+in\s+our\s+(system|database|records)`), + regexp.MustCompile(`(?i)(do\s+not|don't)\s+have\s+(any\s+)?records?\s+(of|for|about)\s+(you|your)`), + regexp.MustCompile(`(?i)no\s+(matching\s+)?record\s+(of\s+)?(a\s+)?report`), + regexp.MustCompile(`(?i)(do\s+not|don't)\s+have\s+(any\s+)?record\s+of\s+(a\s+)?report`), + regexp.MustCompile(`(?i)maintains?\s+no\s+(files?|records?|data)`), + regexp.MustCompile(`(?i)(we\s+)?(have\s+)?no\s+data\s+(linked|associated|related)\s+to\s+(your|this)`), + regexp.MustCompile(`(?i)has\s+never\s+existed\s+in\s+our\s+database`), + regexp.MustCompile(`(?i)(your\s+)?(email|name|address|information)\s+was\s+not\s+identified`), +} + +var subjectNoDataPatterns = []*regexp.Regexp{ + regexp.MustCompile(`(?i)\bno\s+(matching\s+)?(records?|data)(\s+(held|found|stored))?\b`), + regexp.MustCompile(`(?i)(deletion|removal|erasure).{0,60}not\s+found`), + regexp.MustCompile(`(?i)unable\s+to\s+(locate|find)\s+your\s+(data|records?|information|profile)`), +} + +// Conditional statements and requests for proof are not a completed search. +var uncertainNoDataPattern = regexp.MustCompile(`(?i)\b(if|unless|may|might|could|would|si|falls|wenn|peut-être|vielleicht)\b.{0,80}\b(no|not|keine|aucune)\b|\b(cannot|can't|unable\s+to)\s+(locate|find|identify|verify)\b`) +var quotedReplyPattern = regexp.MustCompile(`(?im)^\s*(on .+ wrote:|from:|sent:|de :|von:|le .+écrit\s*:|-{2,}\s*(original message|forwarded message))`) +var blockquotePattern = regexp.MustCompile(`(?is)]*>.*?`) + +func anyPattern(patterns []*regexp.Regexp, text string) bool { + for _, p := range patterns { + if p.MatchString(text) { + return true + } + } + return false +} + +// Only use the broker's reply, not a quoted request asking them to confirm +// that they hold no data. Raw content stays intact in the history database. +func noDataReplyText(content string) string { + content = blockquotePattern.ReplaceAllString(content, "") + if at := quotedReplyPattern.FindStringIndex(content); at != nil { + content = content[:at[0]] + } + var lines []string + for _, line := range strings.Split(content, "\n") { + if !strings.HasPrefix(strings.TrimSpace(line), ">") { + lines = append(lines, line) + } + } + return stripHTML(strings.Join(lines, "\n")) +} + +// noDataVerdict returns a separate, conservative terminal verdict. Generic +// acknowledgments and footer URLs don't outweigh a clear no-records reply; +// competing instructions, a refusal, or a deletion claim require review. +func noDataVerdict(subject, content string) (matched, review bool) { + content = noDataReplyText(content) + bodyMatch := anyPattern(noDataPatterns, content) + matched = bodyMatch || anyPattern(subjectNoDataPatterns, subject) + if !matched { + return false, false + } + // An acknowledgment repeating a no-records subject is not a final answer. + review = (!bodyMatch && (anyPattern(pendingPatterns, content) || anyPattern(subjectPendingPatterns, subject))) || + uncertainNoDataPattern.MatchString(content) || uncertainNoDataPattern.MatchString(subject) || + (content == "" && strings.Contains(strings.ToLower(subject), "not found")) || + anyPattern(formRequiredPatterns, content) || anyPattern(confirmationPatterns, content) || + anyPattern(rejectionPatterns, content) || anyPattern(successPatterns, content) || + anyPattern(subjectRejectionPatterns, subject) || anyPattern(subjectFormPatterns, subject) || + anyPattern(subjectSuccessPatterns, subject) + return matched, review +} diff --git a/internal/inbox/classifier_test.go b/internal/inbox/classifier_test.go index 34f2406..be25b90 100644 --- a/internal/inbox/classifier_test.go +++ b/internal/inbox/classifier_test.go @@ -200,7 +200,7 @@ func TestRejectionPatterns(t *testing.T) { { name: "ACUTRAQ - no record", body: "we do not have any record of a report in our system. ACUTRAQ maintains no files on this person.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "Atlantic Fox - no longer a data broker", @@ -215,7 +215,7 @@ func TestRejectionPatterns(t *testing.T) { { name: "No data linked to email", body: "We have no data linked to your email in our system.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "B2B platform", @@ -225,7 +225,7 @@ func TestRejectionPatterns(t *testing.T) { { name: "Never existed in database", body: "the information below has never existed in our database.", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "FCRA exempt", @@ -235,7 +235,7 @@ func TestRejectionPatterns(t *testing.T) { { name: "Not identified in database", body: "Your email was not identified in our database. This means that we don't have any personal information.", - expected: ResponseRejected, + expected: ResponseNoData, }, } @@ -341,19 +341,19 @@ func TestSubjectBasedRejectionPatterns(t *testing.T) { name: "Not Found in subject", subject: "[Convex] Re: Deletion Request Update - Not Found", body: "", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "No record in subject", subject: "Re: Personal Data Removal Request - No Record Found", body: "", - expected: ResponseRejected, + expected: ResponseNoData, }, { name: "Unable to locate", subject: "Unable to locate your data - Personal Data Removal Request", body: "", - expected: ResponseRejected, + expected: ResponseNoData, }, } diff --git a/internal/inbox/migrate.go b/internal/inbox/migrate.go new file mode 100644 index 0000000..0155f1f --- /dev/null +++ b/internal/inbox/migrate.go @@ -0,0 +1,15 @@ +package inbox + +import "github.com/drumandbytes/eraser/internal/history" + +// MigrateNoDataResponses uses only locally stored messages, never IMAP. +func MigrateNoDataResponses(store *history.Store) error { + return store.MigrateNoData(func(r history.BrokerResponse) (bool, float64, bool) { + if r.EmailBody == "" { + t, confidence, review := ClassifyBySubjectOnly(r.EmailSubject) + return t == ResponseNoData, confidence, review + } + classified := ClassifyResponse(&Email{From: r.EmailFrom, Subject: r.EmailSubject, Body: r.EmailBody}) + return classified.Type == ResponseNoData, classified.Confidence, classified.NeedsReview + }) +} diff --git a/internal/inbox/no_data_test.go b/internal/inbox/no_data_test.go new file mode 100644 index 0000000..ec51c0f --- /dev/null +++ b/internal/inbox/no_data_test.go @@ -0,0 +1,85 @@ +package inbox + +import ( + "path/filepath" + "testing" + "time" + + "github.com/drumandbytes/eraser/internal/history" +) + +func TestNoDataClassification(t *testing.T) { + for _, tc := range []struct { + name, subject, body, html string + kind ResponseType + review bool + }{ + {"clear", "Re: request", "We do not have any records of you in our database.", "", ResponseNoData, false}, + {"no records of you", "Re: request", "We have no records of you in our database.", "", ResponseNoData, false}, + {"no data about you", "Re: request", "We hold no data about you.", "", ResponseNoData, false}, + {"courtesy", "Ticket #123456 received", "We hold no personal data concerning you. If you have questions, contact our team.", "", ResponseNoData, false}, + {"html", "Re: request", "", "

We have no data linked to your email.

", ResponseNoData, false}, + {"subject", "No matching records found", "", "", ResponseNoData, false}, + {"german", "Ihre Anfrage", "Wir speichern keine personenbezogenen Daten über Ihnen.", "", ResponseNoData, false}, + {"french", "Votre demande", "Nous ne détenons aucune donnée vous concernant.", "", ResponseNoData, false}, + {"conditional", "Re: request", "If no records found, we will tell you.", "", ResponseNoData, true}, + {"subject echoed by acknowledgment", "Re: No records found", "We have received your request and will get back to you.", "", ResponseNoData, true}, + {"conflicting subject", "Request completed", "We have no data linked to your email.", "", ResponseNoData, true}, + {"verification", "Re: request", "No matching records found. Please verify your identity before we can proceed.", "", ResponseNoData, true}, + {"mixed refusal", "Re: request", "We hold no personal data concerning you. Your request has been rejected.", "", ResponseNoData, true}, + {"uncertain subject", "Unable to locate your data", "", "", ResponseNoData, true}, + {"deletion", "Re: request", "We have deleted your data and no longer hold your information.", "", ResponseSuccess, false}, + {"refusal", "Request denied", "We cannot process your request.", "", ResponseRejected, false}, + {"quoted request", "Re: request", "We have received your request.\nOn Monday you wrote:\nNo records found", "", ResponsePending, false}, + {"quoted html", "Re: request", "", "

We have received your request.

Please confirm no records found
", ResponsePending, false}, + {"bounce", "Delivery status notification (failure)", "Your message could not be delivered. No records found.", "", ResponseBounced, false}, + } { + t.Run(tc.name, func(t *testing.T) { + got := ClassifyResponse(&Email{Subject: tc.subject, Body: tc.body, HTMLBody: tc.html}) + if got.Type != tc.kind || got.NeedsReview != tc.review { + t.Fatalf("got %s review=%v confidence=%v, want %s review=%v", got.Type, got.NeedsReview, got.Confidence, tc.kind, tc.review) + } + }) + } + if kind, confidence, review := ClassifyBySubjectOnly("No matching records found"); kind != ResponseNoData || confidence < 0.7 || review { + t.Fatal("clear subject-only classification missing") + } +} + +func TestNoDataNewReplyResolvesOnlyConfidentCurrentOutcome(t *testing.T) { + for _, review := range []bool{false, true} { + t.Run(map[bool]string{false: "clear", true: "ambiguous"}[review], func(t *testing.T) { + store, err := history.NewStore(filepath.Join(t.TempDir(), "history.db")) + if err != nil { + t.Fatal(err) + } + defer func() { _ = store.Close() }() + if err := store.Add(&history.Record{BrokerID: "acme", BrokerName: "Acme", Status: history.StatusSent, SentAt: time.Now().Add(-time.Hour)}); err != nil { + t.Fatal(err) + } + body := "We do not have any records of you." + if review { + body += " Please verify your identity." + } + _, _, err = RecordReply(store, &Email{BrokerID: "acme", BrokerName: "Acme", Subject: "Our response", Body: body, ReceivedAt: time.Now()}, false) + if err != nil { + t.Fatal(err) + } + requests, _ := store.LatestRequests("default") + want := history.PipelineNoData + if review { + want = history.PipelineEmailSent + } + if requests[0].PipelineStatus != want { + t.Fatalf("pipeline=%s want=%s", requests[0].PipelineStatus, want) + } + }) + } +} + +func TestNoDataReviewKeepsActionLinks(t *testing.T) { + got := ClassifyResponse(&Email{Subject: "Your privacy request", Body: "No records found. Please verify your identity: https://example.com/confirm?token=123"}) + if got.Type != ResponseNoData || !got.NeedsReview || got.ConfirmURL == "" { + t.Fatalf("review should retain verification link: %+v", got) + } +} diff --git a/internal/inbox/scan.go b/internal/inbox/scan.go index c9de7fa..524d8eb 100644 --- a/internal/inbox/scan.go +++ b/internal/inbox/scan.go @@ -121,6 +121,11 @@ func RecordReply(store *history.Store, e *Email, reclassify bool) (ClassifiedRes return classified, ReplyUpdated, err } } + if classified.Type == ResponseNoData { + if err := store.ResolveNoData(profileID, e.BrokerID, true); err != nil { + return classified, ReplyUpdated, err + } + } return classified, ReplyUpdated, nil } } @@ -139,11 +144,23 @@ func RecordReply(store *history.Store, e *Email, reclassify bool) (ClassifiedRes NeedsReview: classified.NeedsReview, ReceivedAt: e.ReceivedAt, }) - if err != nil || !inserted { + if err != nil { + return classified, ReplySeen, err + } + if !inserted { + if classified.Type == ResponseNoData { + err = store.ResolveNoData(profileID, e.BrokerID, true) + } return classified, ReplySeen, err } // No matching request row is fine: the reply still counts. - _ = store.UpdatePipelineStatus(profileID, e.BrokerID, pipelineStatusFor(classified.Type)) + if classified.Type == ResponseNoData { + if err := store.ResolveNoData(profileID, e.BrokerID, false); err != nil { + return classified, ReplyNew, err + } + } else { + _ = store.UpdatePipelineStatus(profileID, e.BrokerID, pipelineStatusFor(classified.Type)) + } return classified, ReplyNew, nil } @@ -151,6 +168,8 @@ func pipelineStatusFor(t ResponseType) history.PipelineStatus { switch t { case ResponseSuccess: return history.PipelineConfirmed + case ResponseNoData: + return history.PipelineNoData case ResponseFormRequired: return history.PipelineFormRequired case ResponseConfirmationRequired: diff --git a/internal/web/handlers_api.go b/internal/web/handlers_api.go index 52ac783..5b6afd6 100644 --- a/internal/web/handlers_api.go +++ b/internal/web/handlers_api.go @@ -174,16 +174,21 @@ func (s *Server) handleAPIDeleteAllHistory(w http.ResponseWriter, r *http.Reques } func (s *Server) handleAPIResponses(w http.ResponseWriter, r *http.Request) { - responseType := r.URL.Query().Get("type") - needsReview := r.URL.Query().Get("needs_review") == "true" - - var responses []history.BrokerResponse - if s.historyStore != nil { - responses, _ = s.historyStore.GetBrokerResponses(s.activeProfile(r).ID, responseType, needsReview, 50) + if s.historyStore == nil { + http.Error(w, "Unable to load responses", http.StatusInternalServerError) + return + } + f := readPipelineFilter(r) + responses, total, err := s.historyStore.QueryBrokerResponses(s.activeProfile(r).ID, history.ResponseQuery{ + Type: r.URL.Query().Get("type"), NeedsReview: r.URL.Query().Get("needs_review") == "true", Search: f.Search, Limit: pipelinePageSize, Offset: (f.Page - 1) * pipelinePageSize, + }) + if err != nil { + http.Error(w, "Unable to load responses", http.StatusInternalServerError) + return } - s.renderPartial(w, "partials/response-list.html", map[string]interface{}{ - "Responses": responses, + "Responses": responses, "Search": f.Search, "Total": total, + "Filtered": f.Search != "" || r.URL.Query().Get("type") != "" || r.URL.Query().Get("needs_review") == "true", }) } @@ -281,6 +286,7 @@ func (s *Server) scanInbox(w http.ResponseWriter, r *http.Request, opt inbox.Sca
Confirm required: %d
Pending: %d
Rejected: %d
+
No data held: %d
Unknown: %d

@@ -288,7 +294,7 @@ func (s *Server) scanInbox(w http.ResponseWriter, r *http.Request, opt inbox.Sca Refresh page

- `, sum.Total, len(res.New), updated, sum.Success, sum.FormRequired, sum.ConfirmRequired, sum.Pending, sum.Rejected, sum.Unknown) + `, sum.Total, len(res.New), updated, sum.Success, sum.FormRequired, sum.ConfirmRequired, sum.Pending, sum.Rejected, sum.NoData, sum.Unknown) } // writeScanAlert writes an .alert- banner (layout.html); detail is escaped. @@ -403,7 +409,7 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) { // Reclassify each response - use full classifier if body available, otherwise subject-only var updated, unchanged int - var pending, rejected, success, formRequired, confirmRequired, unknown int + var pending, rejected, noData, success, formRequired, confirmRequired, unknown int for _, resp := range responses { var newType inbox.ResponseType @@ -431,8 +437,8 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) { confirmURL = resp.ConfirmURL } - // Only update if classification changed or was unknown - if string(newType) != resp.ResponseType || (resp.ResponseType == "unknown" && newType != inbox.ResponseUnknown) { + // Also refresh confidence/review flags for existing no-data outcomes. + if string(newType) != resp.ResponseType || (newType == inbox.ResponseNoData && (confidence != resp.Confidence || needsReview != resp.NeedsReview)) { err := s.historyStore.UpdateBrokerResponseClassification( resp.ID, resp.ProfileID, @@ -442,13 +448,22 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) { confidence, needsReview, ) - if err == nil { - updated++ + if err != nil { + writeScanAlert(w, "error", "Failed to update response:", err.Error()) + return } + updated++ } else { unchanged++ } + if newType == inbox.ResponseNoData { + if err := s.historyStore.ResolveNoData(resp.ProfileID, resp.BrokerID, true); err != nil { + writeScanAlert(w, "error", "Failed to reconcile response:", err.Error()) + return + } + } + // Count by final type switch newType { case inbox.ResponseSuccess: @@ -457,6 +472,8 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) { formRequired++ case inbox.ResponseConfirmationRequired: confirmRequired++ + case inbox.ResponseNoData: + noData++ case inbox.ResponseRejected: rejected++ case inbox.ResponsePending: @@ -476,6 +493,7 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) {
Pending: %d
Rejected: %d
+
No data held: %d
Success: %d
Form required: %d
Confirm required: %d
@@ -486,5 +504,5 @@ func (s *Server) handleAPIReclassify(w http.ResponseWriter, r *http.Request) { Refresh page

- `, len(responses), updated, unchanged, pending, rejected, success, formRequired, confirmRequired, unknown) + `, len(responses), updated, unchanged, pending, rejected, noData, success, formRequired, confirmRequired, unknown) } diff --git a/internal/web/handlers_pages.go b/internal/web/handlers_pages.go index f041b0e..6de380d 100644 --- a/internal/web/handlers_pages.go +++ b/internal/web/handlers_pages.go @@ -101,6 +101,7 @@ type PipelineStats struct { CaptchaSolved int AwaitingConfirmation int Confirmed int + NoData int Rejected int Failed int PendingTasks int @@ -108,81 +109,10 @@ type PipelineStats struct { } func (s *Server) getPipelineStats(profileID string) PipelineStats { - stats := PipelineStats{} - - if s.historyStore == nil { - return stats - } - - pipelineStats, err := s.historyStore.GetPipelineStats(profileID) - if err == nil { - stats.EmailSent = pipelineStats[history.PipelineEmailSent] - stats.AwaitingResponse = pipelineStats[history.PipelineAwaitingResponse] - stats.FormRequired = pipelineStats[history.PipelineFormRequired] - stats.FormFilled = pipelineStats[history.PipelineFormFilled] - stats.AwaitingCaptcha = pipelineStats[history.PipelineAwaitingCaptcha] - stats.CaptchaSolved = pipelineStats[history.PipelineCaptchaSolved] - stats.AwaitingConfirmation = pipelineStats[history.PipelineAwaitingConfirmation] - stats.Confirmed = pipelineStats[history.PipelineConfirmed] - stats.Rejected = pipelineStats[history.PipelineRejected] - stats.Failed = pipelineStats[history.PipelineFailed] - } - - pendingTaskCount, _, _, err := s.historyStore.GetPendingTaskStats(profileID) - if err == nil { - stats.PendingTasks = pendingTaskCount - } - - responses, err := s.historyStore.GetBrokerResponses(profileID, "", true, 1000) - if err == nil { - stats.NeedsReview = len(responses) - } - - // Get form stats (what's actually shown on tasks page) - pendingForms, _, _, _, _, _ := s.historyStore.GetFormStats(profileID) - - // Calculate unified "Action Needed" based on what's displayed on /tasks page: - // - Pending forms (forms without tasks yet) - // - Pending tasks (from pending_tasks table) - // - Items needing review (parser was unsure) - stats.PendingTasks = pendingForms + pendingTaskCount + stats.NeedsReview - + stats, _, _ := s.loadPipelineStats(profileID) return stats } -func (s *Server) handlePipeline(w http.ResponseWriter, r *http.Request) { - cfg := s.getConfig() - if cfg == nil || !cfg.HasProfile() { - http.Redirect(w, r, "/setup", http.StatusFound) - return - } - - active := s.activeProfile(r) - pipelineStats := s.getPipelineStats(active.ID) - - // Get recent responses - var recentResponses []history.BrokerResponse - if s.historyStore != nil { - recentResponses, _ = s.historyStore.GetBrokerResponses(active.ID, "", false, 20) - } - - // Get pending tasks - var pendingTasks []history.PendingTask - if s.historyStore != nil { - pendingTasks, _ = s.historyStore.GetPendingTasks(active.ID, "", "pending") - } - - data := map[string]interface{}{ - "Title": "Pipeline Status", - "PipelineStats": pipelineStats, - "Responses": recentResponses, - "PendingTasks": pendingTasks, - "InboxConfigured": cfg.InboxForProfile(active).Enabled, - } - - s.renderWithCSRF(w, r, "pipeline.html", data) -} - func (s *Server) handleResponseReview(w http.ResponseWriter, r *http.Request) { var responseID int64 if _, err := fmt.Sscanf(chi.URLParam(r, "responseID"), "%d", &responseID); err != nil { @@ -269,55 +199,54 @@ func (s *Server) handleTasks(w http.ResponseWriter, r *http.Request) { status = "pending" } - var tasks []history.PendingTask - var completedTasksList []history.PendingTask + var actions history.ActionItems var forms []history.FormWithStatus - var reviewItems []history.BrokerResponse - if s.historyStore != nil { - tasks, _ = s.historyStore.GetPendingTasks(active.ID, history.TaskType(taskType), "pending") - completedTasksList, _ = s.historyStore.GetPendingTasks(active.ID, history.TaskType(taskType), "completed") - forms, _ = s.historyStore.GetFormsWithStatus(active.ID) - reviewItems, _ = s.historyStore.GetBrokerResponses(active.ID, "", true, 1000) - } - - pendingTasks, completedTasksCount, skippedTasks := 0, 0, 0 - if s.historyStore != nil { - pendingTasks, completedTasksCount, skippedTasks, _ = s.historyStore.GetPendingTaskStats(active.ID) - } - - pendingForms, filledForms, captchaForms, failedForms, skippedForms := 0, 0, 0, 0, 0 + var completedTasksList []history.PendingTask + var completedTasksCount, skippedTasks, filledForms, captchaForms, failedForms, skippedForms int if s.historyStore != nil { - pendingForms, filledForms, captchaForms, failedForms, skippedForms, _ = s.historyStore.GetFormStats(active.ID) + var err error + actions, err = s.historyStore.GetActionItems(active.ID) + if err != nil { + http.Error(w, "Unable to load action items", http.StatusInternalServerError) + return + } + forms, err = s.historyStore.GetFormsWithStatus(active.ID) + if err != nil { + http.Error(w, "Unable to load forms", http.StatusInternalServerError) + return + } + // Pending forms come exclusively from the shared action projection. + kept := append([]history.FormWithStatus{}, actions.Forms...) + for _, f := range forms { + if f.Status != "pending" { + kept = append(kept, f) + } + } + forms = kept + completedTasksList, _ = s.historyStore.GetPendingTasks(active.ID, "", "completed") + _, completedTasksCount, skippedTasks, _ = s.historyStore.GetPendingTaskStats(active.ID) + _, filledForms, captchaForms, failedForms, skippedForms, _ = s.historyStore.GetFormStats(active.ID) } - // Count forms needing action (only pending forms, not captcha since those are in pendingTasks) - formsNeedingAction := pendingForms - - // Total action items: pending forms (without tasks) + pending tasks - // This avoids double-counting captcha items - totalActionItems := pendingForms + pendingTasks - - needsReviewCount := len(reviewItems) - data := map[string]interface{}{ "Title": "Action Needed", - "Tasks": tasks, + "Tasks": actions.Tasks, "CompletedTasksList": completedTasksList, "Forms": forms, - "ReviewItems": reviewItems, + "ReviewItems": actions.Reviews, "TaskType": taskType, "Status": status, - "PendingTasks": pendingTasks, + "PendingTasks": len(actions.Tasks), "CompletedTasks": completedTasksCount, "SkippedTasks": skippedTasks, - "PendingForms": pendingForms, + "PendingForms": len(actions.Forms), "FilledForms": filledForms, "CaptchaForms": captchaForms, "FailedForms": failedForms, "SkippedForms": skippedForms, - "NeedsReview": needsReviewCount, - "FormsNeedingAction": formsNeedingAction, - "TotalActionItems": totalActionItems + needsReviewCount, + "NeedsReview": actions.NeedsReview(), + "FormsNeedingAction": actions.FormsToFill(), + "TotalActionItems": actions.Total(), } s.renderWithCSRF(w, r, "tasks.html", data) diff --git a/internal/web/handlers_pages_test.go b/internal/web/handlers_pages_test.go index 71745d6..9c3f6f6 100644 --- a/internal/web/handlers_pages_test.go +++ b/internal/web/handlers_pages_test.go @@ -1,8 +1,10 @@ package web import ( + "github.com/drumandbytes/eraser/internal/history" "net/http" "net/http/httptest" + "path/filepath" "strings" "testing" @@ -23,6 +25,12 @@ func TestPipelineShowsScanButtonForProfileMailOverride(t *testing.T) { // profile's own override, not a shared default. }) + store, err := history.NewStore(filepath.Join(t.TempDir(), "history.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + s.historyStore = store req := httptest.NewRequest(http.MethodGet, "/pipeline", nil) req.AddCookie(&http.Cookie{Name: activeProfileCookie, Value: "spouse"}) w := httptest.NewRecorder() diff --git a/internal/web/pipeline.go b/internal/web/pipeline.go new file mode 100644 index 0000000..e80bbe6 --- /dev/null +++ b/internal/web/pipeline.go @@ -0,0 +1,358 @@ +package web + +import ( + "fmt" + "io" + "net/http" + "net/url" + "sort" + "strconv" + "strings" + "time" + + "github.com/drumandbytes/eraser/internal/history" + "golang.org/x/net/html" +) + +const pipelinePageSize = 50 + +type pipelineFilter struct { + View, Status, Type, Search string + NeedsReview bool + Page int +} + +func readPipelineFilter(r *http.Request) pipelineFilter { + q := r.URL.Query() + f := pipelineFilter{View: q.Get("view"), Status: q.Get("status"), Type: q.Get("type"), Search: strings.TrimSpace(q.Get("q")), NeedsReview: q.Get("needs_review") == "true"} + if f.View == "" && f.Status != "" { + f.View = "requests" + } + if f.View != "responses" && f.View != "requests" { + f.View = "actions" + } + if f.View != "responses" { + f.Type = "" + } + if f.View != "requests" { + f.Status = "" + } + if f.View == "requests" { + f.NeedsReview = false + } + f.Page, _ = strconv.Atoi(q.Get("page")) + if f.Page < 1 || f.Page > 1000000 { + f.Page = 1 + } + return f +} + +func (f pipelineFilter) URL() string { + q := url.Values{"view": {f.View}} + if f.Status != "" { + q.Set("status", f.Status) + } + if f.Type != "" { + q.Set("type", f.Type) + } + if f.Search != "" { + q.Set("q", f.Search) + } + if f.NeedsReview { + q.Set("needs_review", "true") + } + if f.Page > 1 { + q.Set("page", strconv.Itoa(f.Page)) + } + return "/pipeline?" + q.Encode() +} + +type pipelineItem struct { + BrokerName, BrokerID, Kind, Label, Subject, Snippet, URL, Action, FormURL string + At time.Time + ID int64 + NeedsReview bool +} +type pipelineCard struct { + Label, URL string + Count int + Active bool +} +type pipelineView struct { + Filter pipelineFilter + Cards []pipelineCard + Items []pipelineItem + Total, Start, End, Pages int + PreviousURL, NextURL, ActionsURL, ResponsesURL, RequestsURL, ClearURL string + Heading, LoadError string +} + +func pipelineLabel(kind string) string { + labels := map[string]string{ + "email_sent": "Email sent", "awaiting_response": "Awaiting response", "form_required": "Form required", + "form_filled": "Form filled", "awaiting_captcha": "Awaiting CAPTCHA", "captcha_solved": "CAPTCHA solved", + "awaiting_confirmation": "Awaiting confirmation", "confirmed": "Confirmed", "success": "Deletion confirmed", + "no_data": "No data held", "rejected": "Rejected", "failed": "Failed", "pending": "Processing", + "unknown": "Unknown", "bounced": "Bounced", "manual_form": "Form required", "form": "Form required", + "confirmation_required": "Confirmation required", "confirm": "Confirmation required", "confirmation": "Confirmation required", + "captcha": "CAPTCHA", "review": "Needs review", + } + if label, ok := labels[kind]; ok { + return label + } + return strings.ReplaceAll(kind, "_", " ") +} + +// responsePreview is plain text, still escaped by html/template. Matching +// snippets let a body-only search explain why a reply was returned. +func responsePreview(body, search string) string { + z := html.NewTokenizer(strings.NewReader(body)) + var b strings.Builder + skip := 0 + for { + tt := z.Next() + if tt == html.ErrorToken { + if z.Err() == io.EOF { + break + } + break + } + if tt == html.StartTagToken || tt == html.EndTagToken { + name, _ := z.TagName() + if string(name) == "script" || string(name) == "style" { + if tt == html.StartTagToken { + skip++ + } else if skip > 0 { + skip-- + } + } + b.WriteByte(' ') + } + if tt == html.TextToken && skip == 0 { + b.Write(z.Text()) + } + } + text := strings.Join(strings.Fields(b.String()), " ") + runes := []rune(text) + start := 0 + if search != "" { + lowered := strings.ToLower(text) + if at := strings.Index(lowered, strings.ToLower(search)); at > 0 { + start = max(0, len([]rune(lowered[:at]))-55) + } + } + end := min(len(runes), start+220) + preview := string(runes[start:end]) + if start > 0 { + preview = "…" + preview + } + if end < len(runes) { + preview += "…" + } + return preview +} + +func (s *Server) loadPipelineStats(profileID string) (PipelineStats, history.ActionItems, error) { + var stats PipelineStats + var actions history.ActionItems + if s.historyStore == nil { + return stats, actions, fmt.Errorf("database not available") + } + stages, err := s.historyStore.GetPipelineStats(profileID) + if err != nil { + return stats, actions, err + } + actions, err = s.historyStore.GetActionItems(profileID) + if err != nil { + return stats, actions, err + } + stats = PipelineStats{ + EmailSent: stages[history.PipelineEmailSent], AwaitingResponse: stages[history.PipelineAwaitingResponse], + FormRequired: stages[history.PipelineFormRequired], FormFilled: stages[history.PipelineFormFilled], + AwaitingCaptcha: stages[history.PipelineAwaitingCaptcha], CaptchaSolved: stages[history.PipelineCaptchaSolved], + AwaitingConfirmation: stages[history.PipelineAwaitingConfirmation], Confirmed: stages[history.PipelineConfirmed], + NoData: stages[history.PipelineNoData], Rejected: stages[history.PipelineRejected], Failed: stages[history.PipelineFailed], + PendingTasks: actions.Total(), NeedsReview: actions.NeedsReview(), + } + return stats, actions, nil +} + +func newPipelineView(f pipelineFilter, stats PipelineStats) pipelineView { + v := pipelineView{Filter: f, Heading: "Action Required"} + for _, card := range []struct { + label, status string + count int + }{ + {"Email Sent", "email_sent", stats.EmailSent}, {"Awaiting Response", "awaiting_response", stats.AwaitingResponse}, + {"Form Required", "form_required", stats.FormRequired}, {"Awaiting Confirm", "awaiting_confirmation", stats.AwaitingConfirmation}, + {"Confirmed", "confirmed", stats.Confirmed}, {"No data held", "no_data", stats.NoData}, {"Rejected", "rejected", stats.Rejected}, + {"Pending Tasks", "actions", stats.PendingTasks}, {"Needs Review", "review", stats.NeedsReview}, + } { + cf := pipelineFilter{View: "requests", Status: card.status, Search: f.Search, Page: 1} + active := f.View == "requests" && f.Status == card.status + if card.status == "actions" || card.status == "review" { + cf.View, cf.Status, cf.NeedsReview = "actions", "", card.status == "review" + active = f.View == "actions" && f.NeedsReview == cf.NeedsReview + } + v.Cards = append(v.Cards, pipelineCard{card.label, cf.URL(), card.count, active}) + } + for _, name := range []string{"actions", "responses", "requests"} { + u := (pipelineFilter{View: name, Search: f.Search, Page: 1}).URL() + switch name { + case "actions": + v.ActionsURL = u + case "responses": + v.ResponsesURL = u + case "requests": + v.RequestsURL = u + } + } + clear := f + clear.Search = "" + clear.Type = "" + clear.Status = "" + clear.NeedsReview = false + clear.Page = 1 + v.ClearURL = clear.URL() + if f.View == "responses" { + v.Heading = "Responses" + } + if f.View == "requests" { + v.Heading = "Requests" + if f.Status != "" { + v.Heading = pipelineLabel(f.Status) + } + } + if f.NeedsReview { + v.Heading = "Needs Review" + } + return v +} + +func (s *Server) pipelineResults(profileID string, f pipelineFilter, stats PipelineStats, actions history.ActionItems) (pipelineView, error) { + v := newPipelineView(f, stats) + if s.historyStore == nil { + return v, fmt.Errorf("database not available") + } + if f.View == "responses" { + responses, total, err := s.historyStore.QueryBrokerResponses(profileID, history.ResponseQuery{Type: f.Type, NeedsReview: f.NeedsReview, Search: f.Search, Limit: pipelinePageSize, Offset: (f.Page - 1) * pipelinePageSize}) + if err != nil { + return v, err + } + v.Total = total + if f.Page > 1 && (f.Page-1)*pipelinePageSize >= total { + f.Page = 1 + return s.pipelineResults(profileID, f, stats, actions) + } + for _, r := range responses { + v.Items = append(v.Items, pipelineItem{BrokerName: r.BrokerName, BrokerID: r.BrokerID, Kind: r.ResponseType, Label: pipelineLabel(r.ResponseType), Subject: r.EmailSubject, Snippet: responsePreview(r.EmailBody, f.Search), URL: fmt.Sprintf("/pipeline/responses/%d", r.ID), Action: "Read response", At: r.ReceivedAt, ID: r.ID, NeedsReview: r.NeedsReview}) + } + } else { + matches := func(parts ...string) bool { + return f.Search == "" || strings.Contains(strings.ToLower(strings.Join(parts, " ")), strings.ToLower(f.Search)) + } + if f.View == "requests" { + requests, err := s.historyStore.LatestRequests(profileID) + if err != nil { + return v, err + } + for _, r := range requests { + if (f.Status != "" && string(r.PipelineStatus) != f.Status) || !matches(r.BrokerName, r.BrokerID, r.Email) { + continue + } + v.Items = append(v.Items, pipelineItem{BrokerName: r.BrokerName, BrokerID: r.BrokerID, Kind: string(r.PipelineStatus), Label: pipelineLabel(string(r.PipelineStatus)), Subject: r.Email, URL: "/brokers?search=" + url.QueryEscape(r.BrokerName), Action: "View broker", At: r.SentAt, ID: r.ID}) + } + } else { + var matching map[int64]bool + if f.Search != "" { + var err error + matching, err = s.historyStore.MatchingResponseIDs(profileID, f.Search) + if err != nil { + return v, err + } + } + for _, form := range actions.Forms { + if f.NeedsReview || (!matches(form.BrokerName, form.BrokerID, form.EmailSubject) && !matching[form.ResponseID]) { + continue + } + v.Items = append(v.Items, pipelineItem{BrokerName: form.BrokerName, BrokerID: form.BrokerID, Kind: "form", Label: "Form required", Subject: form.EmailSubject, URL: "/tasks#form-" + form.BrokerID, Action: "View form", FormURL: form.FormURL, At: form.DetectedAt, ID: form.ResponseID}) + } + for _, task := range actions.Tasks { + review := task.TaskType == history.TaskReview + if (f.NeedsReview && !review) || !matches(task.BrokerName, task.BrokerID, task.Notes, string(task.TaskType)) { + continue + } + v.Items = append(v.Items, pipelineItem{BrokerName: task.BrokerName, BrokerID: task.BrokerID, Kind: string(task.TaskType), Label: pipelineLabel(string(task.TaskType)), Snippet: responsePreview(task.Notes, f.Search), URL: fmt.Sprintf("/tasks/%d", task.ID), Action: "View task", At: task.CreatedAt, ID: task.ID, NeedsReview: review}) + } + for _, r := range actions.Reviews { + if !matches(r.BrokerName, r.BrokerID, r.EmailSubject, r.EmailFrom) && !matching[r.ID] { + continue + } + v.Items = append(v.Items, pipelineItem{BrokerName: r.BrokerName, BrokerID: r.BrokerID, Kind: r.ResponseType, Label: pipelineLabel(r.ResponseType), Subject: r.EmailSubject, URL: fmt.Sprintf("/pipeline/responses/%d", r.ID), Action: "Review response", At: r.ReceivedAt, ID: r.ID, NeedsReview: true}) + } + } + sort.SliceStable(v.Items, func(i, j int) bool { + a, b := v.Items[i], v.Items[j] + if a.At.Equal(b.At) { + return a.ID > b.ID + } + return a.At.After(b.At) + }) + v.Total = len(v.Items) + if f.Page > 1 && (f.Page-1)*pipelinePageSize >= v.Total { + f.Page = 1 + } + start := (f.Page - 1) * pipelinePageSize + v.Items = v.Items[start:min(start+pipelinePageSize, v.Total)] + } + v.Filter = f + v.Pages = max(1, (v.Total+pipelinePageSize-1)/pipelinePageSize) + if v.Total > 0 { + v.Start = (f.Page-1)*pipelinePageSize + 1 + v.End = v.Start + len(v.Items) - 1 + } + if f.Page > 1 { + prev := f + prev.Page-- + v.PreviousURL = prev.URL() + } + if f.Page < v.Pages { + next := f + next.Page++ + v.NextURL = next.URL() + } + return v, nil +} + +func (s *Server) handlePipeline(w http.ResponseWriter, r *http.Request) { + cfg := s.getConfig() + if cfg == nil || !cfg.HasProfile() { + http.Redirect(w, r, "/setup", http.StatusFound) + return + } + active := s.activeProfile(r) + f := readPipelineFilter(r) + stats, actions, err := s.loadPipelineStats(active.ID) + v := newPipelineView(f, stats) + if err == nil { + v, err = s.pipelineResults(active.ID, f, stats, actions) + } + if err != nil { + v.LoadError = "Unable to load pipeline items. Please try again." + } + w.Header().Add("Vary", "HX-Request") + w.Header().Add("Vary", "HX-History-Restore-Request") + // History restoration needs the full document, not only the list fragment. + if r.Header.Get("HX-Request") == "true" && r.Header.Get("HX-History-Restore-Request") != "true" { + w.Header().Set("HX-Push-Url", v.Filter.URL()) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + } + s.renderPartial(w, "partials/pipeline-content.html", v) + return + } + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + } + s.renderWithCSRF(w, r, "pipeline.html", map[string]interface{}{"Title": "Pipeline", "Pipeline": v, "InboxConfigured": cfg.InboxForProfile(active).Enabled}) +} diff --git a/internal/web/pipeline_test.go b/internal/web/pipeline_test.go new file mode 100644 index 0000000..8de8b29 --- /dev/null +++ b/internal/web/pipeline_test.go @@ -0,0 +1,207 @@ +package web + +import ( + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/drumandbytes/eraser/internal/history" +) + +func pipelineTestServer(t *testing.T) *Server { + t.Helper() + s := newTestServer(t, testConfig("default", "other")) + store, err := history.NewStore(filepath.Join(t.TempDir(), "history.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = store.Close() }) + s.historyStore = store + return s +} + +func pipelineReply(t *testing.T, s *Server, r history.BrokerResponse) history.BrokerResponse { + t.Helper() + if r.ProfileID == "" { + r.ProfileID = "default" + } + if r.ReceivedAt.IsZero() { + r.ReceivedAt = time.Now() + } + if err := s.historyStore.AddBrokerResponse(&r); err != nil { + t.Fatal(err) + } + return r +} + +func pipelineGET(s *Server, path string, hx bool) *httptest.ResponseRecorder { + r := httptest.NewRequest(http.MethodGet, path, nil) + if hx { + r.Header.Set("HX-Request", "true") + } + w := httptest.NewRecorder() + s.handlePipeline(w, r) + return w +} + +func TestPipelineEighteenReviewsWithoutTaskRecords(t *testing.T) { + s := pipelineTestServer(t) + for i := range 18 { + pipelineReply(t, s, history.BrokerResponse{BrokerID: fmt.Sprint(i), BrokerName: fmt.Sprintf("Review broker %02d", i), ResponseType: "pending", NeedsReview: true}) + } + stats := s.getPipelineStats("default") + if stats.PendingTasks != 18 || stats.NeedsReview != 18 { + t.Fatalf("stats=%+v", stats) + } + for _, hx := range []bool{false, true} { + w := pipelineGET(s, "/pipeline", hx) + body := w.Body.String() + if w.Code != 200 || strings.Count(body, `
") { + t.Error("history restore requires full document") + } +} + +func TestPipelineReadFailureIsNotAnEmptyState(t *testing.T) { + s := pipelineTestServer(t) + _ = s.historyStore.Close() + for _, hx := range []bool{false, true} { + w := pipelineGET(s, "/pipeline", hx) + if w.Code != 500 || !strings.Contains(w.Body.String(), "Unable to load pipeline") || strings.Contains(w.Body.String(), "No action required") { + t.Fatal("read failure shown as all clear") + } + } +} + +func TestResponsePreviewIsPlainText(t *testing.T) { + got := responsePreview("

We hold no data.

", "") + if strings.Contains(got, "bad()") || strings.Contains(got, "<") || !strings.Contains(got, "no data") { + t.Fatalf("preview=%q", got) + } +} + +func TestResponsePreviewUnicodeMatch(t *testing.T) { + // Case folding can change the UTF-8 byte length (Ⱥ -> ⱥ). Snippet offsets + // must still use rune positions and never slice beyond the original text. + body := strings.Repeat("Ⱥ", 250) + " ÅSE's records" + if got := responsePreview(body, "åse"); !strings.Contains(got, "ÅSE's records") { + t.Fatalf("preview missed Unicode search: %q", got) + } +} diff --git a/internal/web/server.go b/internal/web/server.go index 5039a04..117d702 100644 --- a/internal/web/server.go +++ b/internal/web/server.go @@ -196,6 +196,7 @@ func limitFormBody(w http.ResponseWriter, r *http.Request) { // Each page gets its own template set to avoid "content" block conflicts func (s *Server) parseTemplates() (map[string]*template.Template, error) { funcs := template.FuncMap{ + "pipelineLabel": pipelineLabel, "formatTime": func(t time.Time) string { return t.Format("Jan 2, 2006 3:04 PM") }, diff --git a/internal/web/templates/dashboard.html b/internal/web/templates/dashboard.html index 55bba2a..00b9317 100644 --- a/internal/web/templates/dashboard.html +++ b/internal/web/templates/dashboard.html @@ -111,7 +111,7 @@

Response Pipeline -
+

{{.PipelineStats.AwaitingResponse}}

@@ -127,8 +127,12 @@

Response Pipeline{{.PipelineStats.Rejected}}

Request Rejected

 + +

{{.PipelineStats.NoData}}

+

No data held

+ - +

{{.PipelineStats.PendingTasks}}

Action Needed

 diff --git a/internal/web/templates/partials/pipeline-content.html b/internal/web/templates/partials/pipeline-content.html new file mode 100644 index 0000000..c4abb41 --- /dev/null +++ b/internal/web/templates/partials/pipeline-content.html @@ -0,0 +1,130 @@ +
+ {{if .LoadError}} + + {{else}} + + {{end}} + +
+ + +
+ +
+ + +
+ {{if eq .Filter.View "responses"}} +
+ + +
+ {{end}} + {{if eq .Filter.View "requests"}} +
+ + +
+ {{else}} + + {{end}} + + Clear filters +
+ +
+

{{.Heading}}

+ {{if not .LoadError}}{{.Total}} result{{if ne .Total 1}}s{{end}}{{end}} + Loading… +
+ +
+ {{range .Items}} +
+
+
+ {{.BrokerName}} + {{.Label}} + {{if .NeedsReview}}Needs review{{end}} +
+ {{if .Subject}}

{{.Subject}}

{{end}} + {{if .Snippet}}

{{.Snippet}}

{{end}} +

{{formatTime .At}}

+
+
+ {{.Action}} + {{if .FormURL}}Open form ↗{{end}} +
+
+ {{else}} + {{if not .LoadError}} +
+ {{if or .Filter.Search .Filter.Type .Filter.Status .Filter.NeedsReview}} +

No matching results

Try another search or clear the filters.

+ {{else if eq .Filter.View "actions"}} +

No action required

Your pending forms, tasks and response reviews will appear here.

+ {{else if eq .Filter.View "responses"}} +

No responses yet

Scan your inbox to check for broker responses.

+ {{else}} +

No requests yet

Sent removal requests will appear here.

+ {{end}} +
+ {{end}} + {{end}} +
+ + {{if gt .Total 0}} + + {{end}} +
+
diff --git a/internal/web/templates/partials/response-list.html b/internal/web/templates/partials/response-list.html index 4997aa5..270d794 100644 --- a/internal/web/templates/partials/response-list.html +++ b/internal/web/templates/partials/response-list.html @@ -12,13 +12,15 @@ {{else if eq .ResponseType "confirmation_required"}} + {{else if eq .ResponseType "no_data"}} + {{else if eq .ResponseType "rejected"}} {{else}} {{end}}
-

{{.BrokerName}}

+ {{.BrokerName}}

{{.EmailSubject}}

@@ -26,9 +28,10 @@ {{if eq .ResponseType "success"}}chip chip-success {{else if eq .ResponseType "form_required"}}chip chip-warning {{else if eq .ResponseType "confirmation_required"}}chip chip-info + {{else if eq .ResponseType "no_data"}}chip chip-info {{else if eq .ResponseType "rejected"}}chip chip-error {{else}}chip chip-warning{{end}}"> - {{.ResponseType}} + {{pipelineLabel .ResponseType}} {{if .NeedsReview}} Review @@ -75,7 +78,7 @@
-

No responses yet

+

{{if .Filtered}}No matching results{{else}}No responses yet{{end}}

Run eraser monitor to check for broker responses

{{end}} diff --git a/internal/web/templates/pipeline.html b/internal/web/templates/pipeline.html index 6f0d83d..60591c3 100644 --- a/internal/web/templates/pipeline.html +++ b/internal/web/templates/pipeline.html @@ -104,283 +104,45 @@

Removal Pipeline

{{end}} {{if not .InboxConfigured}} - -
-
-
- - - -
-
-

Inbox Monitoring Not Configured

-

To automatically track broker responses, configure inbox monitoring in Settings.

-

Or add the following to your config.yaml (provider: gmail, proton, fastmail, mailbox-org, posteo, icloud, or set server/port):

- 
inbox:
-  enabled: true
-  provider: fastmail
-  email: you@example.org
-  password: your-app-password
-
-
+
+

To fetch new broker responses, configure inbox monitoring in Settings.

{{end}} - -
- -
-
-
- - - -
-
-

Email Sent

-

{{.PipelineStats.EmailSent}}

-
-
-
- - - -
-
- - - -
-
-

Form Required

-

{{.PipelineStats.FormRequired}}

-
-
- - - -
-
-
- - - -
-
-

Awaiting Confirm

-

{{.PipelineStats.AwaitingConfirmation}}

-
-
-
- - -
-
-
- - - -
-
-

Confirmed

-

{{.PipelineStats.Confirmed}}

-
-
-
- - -
-
-
- - - -
-
-

Pending Tasks

-

{{.PipelineStats.PendingTasks}}

-
-
-
- - -
-
-
- - - -
-
-

Needs Review

-

{{.PipelineStats.NeedsReview}}

-
-
-
-
- - -
- -
-
-

Action Required

- View all -
- - {{if .PendingTasks}} -
- {{range .PendingTasks}} -
- {{if eq .TaskType "captcha"}} -
- - - -
- {{else if eq .TaskType "manual_form"}} -
- - - -
- {{else}} -
- - - - -
- {{end}} -
-

{{.BrokerName}}

-

{{.TaskType}} - {{formatDate .CreatedAt}}

-
- - Solve - -
- {{end}} -
- {{else}} -
-
- - - -
-

No pending tasks

-

All clear! No manual intervention needed.

-
- {{end}} -
- - -
-
-

Recent Responses

-
- -
-
- -
- {{template "partials/response-list.html" .}} -
-
-
- - -
-

Pipeline Flow

-
-
- -
-
- - - -
-

Email Sent

-

{{.PipelineStats.EmailSent}}

-
-
→
- - -
-
- - - -
-

Awaiting

-

{{.PipelineStats.AwaitingResponse}}

-
-
→
- - -
-
- - - -
-

Form Required

-

{{.PipelineStats.FormRequired}}

-
-
→
- - -
-
- - - -
-

Confirm Link

-

{{.PipelineStats.AwaitingConfirmation}}

-
-
→
- - -
-
- - - -
-

Confirmed

-

{{.PipelineStats.Confirmed}}

-
-
-
-
+ {{template "partials/pipeline-content.html" .Pipeline}}
- + {{end}} - {{template "layout" .}} diff --git a/internal/web/templates/response-review.html b/internal/web/templates/response-review.html index d674c26..045624a 100644 --- a/internal/web/templates/response-review.html +++ b/internal/web/templates/response-review.html @@ -54,7 +54,7 @@

Classification<
Response type
-
{{.Response.ResponseType}}
+
{{pipelineLabel .Response.ResponseType}}
Confidence
diff --git a/internal/web/templates/tasks.html b/internal/web/templates/tasks.html index 7d726f0..0787d99 100644 --- a/internal/web/templates/tasks.html +++ b/internal/web/templates/tasks.html @@ -89,7 +89,7 @@

Action Needed

Forms - {{.PendingForms}} + {{.FormsNeedingAction}}