diff --git a/internal/httpapi/gmail_handlers.go b/internal/httpapi/gmail_handlers.go index ef86e7b..cd7cd48 100644 --- a/internal/httpapi/gmail_handlers.go +++ b/internal/httpapi/gmail_handlers.go @@ -360,8 +360,19 @@ func (s *Server) SyncGmail(ctx context.Context, user *store.User) (int, error) { Description: msg.Snippet + "\n\n" + msg.Link, Priority: 4, CreatedBy: user.ID, + // The thread, not the message: a reply arriving later is the same + // conversation, and a second task for it would be the duplicate this + // key exists to prevent. + SourceKey: "gmail:thread:" + msg.ThreadID, } if err := s.db.CreateTask(ctx, task, nil); err != nil { + if errors.Is(err, store.ErrDuplicate) { + // Somebody got there first — an earlier sweep, or another + // assistant reading the same mailbox. Mark it seen and move on; + // it is handled, which is all this loop wanted. + seen[id] = true + continue + } s.log.Warn("gmail create task", "err", err) continue } diff --git a/internal/httpapi/mailbox_handlers.go b/internal/httpapi/mailbox_handlers.go index a341e25..d9946db 100644 --- a/internal/httpapi/mailbox_handlers.go +++ b/internal/httpapi/mailbox_handlers.go @@ -202,8 +202,20 @@ func (s *Server) SyncMailbox(ctx context.Context, user *store.User, m *store.Mai Description: msg.Snippet, Priority: 4, CreatedBy: user.ID, + // Host and uid together: a uid is only unique within one mailbox on + // one server, and two accounts will hand out the same small numbers. + SourceKey: fmt.Sprintf("imap:%s:%s:%d", m.Host, m.Folder, msg.UID), } if err := s.db.CreateTask(ctx, task, nil); err != nil { + if errors.Is(err, store.ErrDuplicate) { + // Already a task. Still count it as read, or last_uid never moves + // past it and every sweep from here on stops at the same message. + done = append(done, msg.UID) + if msg.UID > highest { + highest = msg.UID + } + continue + } s.log.Warn("mailbox create task", "err", err, "mailbox", m.ID) continue } diff --git a/internal/store/migrations/0028_kvasir.sql b/internal/store/migrations/0028_kvasir.sql new file mode 100644 index 0000000..ff48548 --- /dev/null +++ b/internal/store/migrations/0028_kvasir.sql @@ -0,0 +1,79 @@ +-- Kvasir: what an assistant did, and what it acted on. +-- +-- Two things, and they are separate on purpose: one is about the outside world, +-- and would be worth having even if Kvasir never existed. + +-- 1. Where a task came from, out there. +-- +-- Nothing today can answer "has this mail already become a task?". A second +-- reader of the same mailbox — another assistant, a rerun of the same sweep, +-- a person — makes a second task, and the only sign is the duplicate itself. +-- That has happened in practice, twice in one run. +-- +-- Namespaced rather than a bare id: `gmail:thread:18f2c…`, `imap::`, +-- `ics:`. A Gmail thread id and an IMAP uid look nothing alike until the +-- day they collide, and a namespace costs one colon. It also means the next +-- source needs no column — which is the point of a key rather than a foreign +-- key. There is no messages table to point at, and this has to work without one. +ALTER TABLE tasks ADD COLUMN source_key TEXT NOT NULL DEFAULT ''; + +-- The index is the interlock, not an optimisation. Two agents reading the same +-- inbox must be unable to make two tasks, and "must be unable" is a constraint +-- rather than a convention somebody remembers to follow. +-- +-- Scoped to the author: two people on one instance connect their own mailboxes, +-- and the same thread reaching both of them is two tasks, correctly. Partial +-- because `created_by` is ON DELETE SET NULL — a task outlives its author, and +-- rows that have lost theirs must not start colliding with each other. +CREATE UNIQUE INDEX tasks_once_per_source + ON tasks (created_by, source_key) + WHERE source_key <> '' AND created_by IS NOT NULL; + +-- 2. Kvasir's ledger. +-- +-- The dial that lets an assistant act on its own is only bearable because of +-- this table. Two questions have to be answerable at any moment: what did it do +-- today, and how do I undo it. Neither is answerable from the tasks themselves. +-- +-- A suggestion and a completed action are the SAME ROW in different states. That +-- is the whole design: "suggest" and "act" are not two code paths, they are one +-- path that writes a different `state`. Accepting a suggestion moves it to done; +-- undoing an action moves it to undone. Nothing has to be written twice, and the +-- feed reads the same either way. +-- +-- `before_json` is the undo. Not a diff and not a description — the fields as +-- they were, so putting them back needs no reasoning about what changed. It is +-- also the honest record: a task edited by hand afterwards will not match, and +-- an undo that would overwrite newer work should say so rather than win. +CREATE TABLE kvasir_actions ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL REFERENCES users (id) ON DELETE CASCADE, + + -- 'file', 'rewrite', 'split', 'nudge', 'prioritise'. Free text rather than a + -- CHECK: a new capability must not need a migration to be recorded, and an + -- unknown kind in the feed is a row nobody rendered, not a broken database. + kind TEXT NOT NULL, + task_id TEXT REFERENCES tasks (id) ON DELETE CASCADE, + + -- One sentence, written when the action was decided. Stored rather than + -- rebuilt: it says what Kvasir believed at the time, and the project it + -- names may since have been renamed or deleted. + summary TEXT NOT NULL, + -- Why it thought so. Shown on request, because a suggestion you cannot + -- interrogate is one you either trust blindly or ignore entirely. + reason TEXT NOT NULL DEFAULT '', + + before_json TEXT NOT NULL DEFAULT '{}', + after_json TEXT NOT NULL DEFAULT '{}', + + -- suggested → done | dismissed, and done → undone. + state TEXT NOT NULL CHECK (state IN ('suggested', 'done', 'dismissed', 'undone')), + + created_at INTEGER NOT NULL, + decided_at INTEGER NOT NULL DEFAULT 0 +); + +-- The feed asks "what is waiting for me, newest first", and the corrections +-- Kvasir learns from ask "what did I get wrong lately" — both read by user and +-- order by time. +CREATE INDEX kvasir_actions_by_user ON kvasir_actions (user_id, created_at DESC); diff --git a/internal/store/source_test.go b/internal/store/source_test.go new file mode 100644 index 0000000..19cad57 --- /dev/null +++ b/internal/store/source_test.go @@ -0,0 +1,109 @@ +package store + +import ( + "context" + "errors" + "testing" +) + +// One mail, one task — however many readers there are. +// +// Nothing could answer "has this already become a task?" before, so a second +// reader of the same mailbox made a second task. That is not hypothetical: a +// briefing run produced "Betal faktura fra Browns" twice in one pass, and a +// separate assistant reading the same inbox does the same thing every sweep. +func TestOneMailBecomesOneTask(t *testing.T) { + db := openTest(t) + ctx := context.Background() + + user := &User{Email: "kristian@example.dk", Name: "Kristian", PasswordHash: "x"} + if err := db.CreateUser(ctx, user, "Indbakke"); err != nil { + t.Fatal(err) + } + inbox := newProject(t, db, user.ID) + + const thread = "gmail:thread:18f2cabc" + first := &Task{ProjectID: inbox, Content: "Svar Mette", CreatedBy: user.ID, SourceKey: thread} + if err := db.CreateTask(ctx, first, nil); err != nil { + t.Fatalf("first task: %v", err) + } + + // The same thread again — a rerun, or somebody else's assistant. + second := &Task{ProjectID: inbox, Content: "Svar Mette", CreatedBy: user.ID, SourceKey: thread} + err := db.CreateTask(ctx, second, nil) + if !errors.Is(err, ErrDuplicate) { + t.Fatalf("second task for the same thread: err = %v, want ErrDuplicate", err) + } + + // And the question that makes the other agent safe. + id, err := db.TaskIDBySource(ctx, user.ID, thread) + if err != nil { + t.Fatal(err) + } + if id != first.ID { + t.Errorf("TaskIDBySource = %q, want %q", id, first.ID) + } + if id, err := db.TaskIDBySource(ctx, user.ID, "gmail:thread:aldrig-set"); err != nil || id != "" { + t.Errorf("an unseen thread answered %q, %v", id, err) + } +} + +// A task somebody typed has no source, and there can be any number of those. +// The index is partial for exactly this reason — without the WHERE clause the +// second hand-written task in the database would be refused. +func TestTasksWithoutASourceDoNotCollide(t *testing.T) { + db := openTest(t) + ctx := context.Background() + + user := &User{Email: "kristian@example.dk", Name: "Kristian", PasswordHash: "x"} + if err := db.CreateUser(ctx, user, "Indbakke"); err != nil { + t.Fatal(err) + } + inbox := newProject(t, db, user.ID) + + for i, name := range []string{"køb kaffe", "ring til Andreas", "hent pakken"} { + if err := db.CreateTask(ctx, &Task{ProjectID: inbox, Content: name, CreatedBy: user.ID}, nil); err != nil { + t.Fatalf("task %d (%s): %v", i, name, err) + } + } + if id, err := db.TaskIDBySource(ctx, user.ID, ""); err != nil || id != "" { + t.Errorf("an empty source key answered %q, %v", id, err) + } +} + +// Two people on one instance, each with their own mailbox. The same thread id +// reaching both of them is two tasks, correctly — the key is scoped to whoever +// the mail arrived for. +func TestTheSameThreadForTwoPeopleIsTwoTasks(t *testing.T) { + db := openTest(t) + ctx := context.Background() + + var ids []string + for _, email := range []string{"kristian@example.dk", "andreas@example.dk"} { + u := &User{Email: email, Name: email, PasswordHash: "x"} + if err := db.CreateUser(ctx, u, "Indbakke"); err != nil { + t.Fatal(err) + } + inbox := newProject(t, db, u.ID) + task := &Task{ProjectID: inbox, Content: "Fælles tråd", CreatedBy: u.ID, + SourceKey: "gmail:thread:delt"} + if err := db.CreateTask(ctx, task, nil); err != nil { + t.Fatalf("%s: %v", email, err) + } + ids = append(ids, task.ID) + } + if ids[0] == ids[1] { + t.Error("the two people got the same task") + } +} + +// A project to hang the tasks on. The tests are about the source key, not about +// where a task lives. +func newProject(t *testing.T, db *DB, ownerID string) string { + t.Helper() + p := &Project{OwnerID: ownerID, Name: "Indbakke"} + if err := db.CreateProject(context.Background(), p); err != nil { + t.Fatalf("create project: %v", err) + } + return p.ID +} diff --git a/internal/store/tasks.go b/internal/store/tasks.go index fb96a69..fc05ad4 100644 --- a/internal/store/tasks.go +++ b/internal/store/tasks.go @@ -32,6 +32,13 @@ type Task struct { // CreatedBy is empty when the author's account has been deleted: `created_by` // is ON DELETE SET NULL so the task survives the person who wrote it. CreatedBy string + // SourceKey names the thing out there this task was made from, namespaced: + // `gmail:thread:18f2c…`. Empty for a task somebody typed. + // + // It exists so that the second reader of a mailbox — another assistant, a + // rerun, a person — can ask "has this already become a task?" and get an + // answer. CreateTask returns ErrDuplicate rather than making a second one. + SourceKey string SortOrder float64 // SubtaskCount and SubtaskDone are what a row can say without being opened: // that there is something underneath it, and how much of it is left. Counted @@ -452,21 +459,48 @@ func (db *DB) CreateTask(ctx context.Context, t *Task, labelNames []string) erro _, err := tx.ExecContext(ctx, ` INSERT INTO tasks (id, project_id, section_id, parent_id, content, description, priority, due_date, due_datetime, due_timezone, duration_min, - recurrence_rule, assignee_id, created_by, sort_order, + recurrence_rule, assignee_id, created_by, source_key, sort_order, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, t.ID, t.ProjectID, nullString(t.SectionID), nullString(t.ParentID), t.Content, t.Description, t.Priority, nullString(t.DueDate), nullTime(t.DueDatetime), nullString(t.DueTimezone), nullInt(t.DurationMin), nullString(t.RecurrenceRule), nullString(t.AssigneeID), nullString(t.CreatedBy), - t.SortOrder, now.Unix(), now.Unix()) + t.SourceKey, t.SortOrder, now.Unix(), now.Unix()) if err != nil { + // The one broken constraint that is not a bug: this source has already + // become a task. The caller skips it rather than making a duplicate, + // which is the entire reason the index exists. + if isUniqueViolation(err) { + return ErrDuplicate + } return err } return setTaskLabels(ctx, tx, t.ID, t.CreatedBy, labelNames) }) } +// TaskIDBySource answers "has this mail already become a task?" — the question +// nothing could answer before, and the one that stops two assistants reading the +// same inbox from making two of everything. Empty id and no error means no. +// +// Deleted tasks count. A task that was made from a mail and then thrown away is +// an answer — remaking it on the next sweep would be the same duplicate wearing +// a different hat. +func (db *DB) TaskIDBySource(ctx context.Context, userID, sourceKey string) (string, error) { + if sourceKey == "" { + return "", nil + } + var id string + err := db.QueryRowContext(ctx, + `SELECT id FROM tasks WHERE created_by = ? AND source_key = ?`, + userID, sourceKey).Scan(&id) + if errors.Is(err, sql.ErrNoRows) { + return "", nil + } + return id, err +} + func (db *DB) nextTaskOrder(ctx context.Context, projectID, sectionID string) (float64, error) { var max sql.NullFloat64 var err error