tasks: key derived captures by external id and record who resolved
A task extracted from mail deduped on the live-norm index only, so once he finished it the row left the live set and the next poll of the same immutable message re-extracted it as a fresh candidate. mavmaild is a read-only reader and marks nothing read, so that repeats forever. Derived rows now carry an ext_id built from the message uid and the extracted span, unique across every status, while voice keeps live-only norm dedupe because saying an errand again is the recurrence signal. A derived source can no longer capture straight to open, and saying a task out loud that Maven had only proposed promotes the candidate instead of answering that it is already in the list. SetTaskStatus was classified AuthRead. Resolving a task is not additive, it erases work off his list, so it is a write, and the row now records the caller that moved it. ListTasks was unbounded. The list-query matcher claimed any utterance with "что мне делать", including "с чем мне помочь", and the urgency stripper matched inside words. Found in review of #60.
This commit is contained in:
kami
+135
-34
@@ -48,16 +48,34 @@ const (
|
||||
//
|
||||
// Due is optional. Weight is an explicit importance hint (0 = none), which the
|
||||
// prioritiser reads; capture never invents one.
|
||||
// ExternalID is the identity of the thing this task was derived FROM — a
|
||||
// message id plus the extracted span, for a source that re-reads the same
|
||||
// immutable text forever. Empty for anything he stated himself.
|
||||
//
|
||||
// ResolvedBy names the caller that moved the task to its terminal state, in the
|
||||
// source vocabulary. Empty while the task is live.
|
||||
type Task struct {
|
||||
ID int64
|
||||
CreatedTs time.Time
|
||||
Text string
|
||||
Source string
|
||||
Evidence string
|
||||
ExternalID string
|
||||
Status string
|
||||
Due *time.Time
|
||||
Weight int
|
||||
ResolvedTs *time.Time
|
||||
ResolvedBy string
|
||||
}
|
||||
|
||||
// CaptureResult — what CaptureTask did. Created is a new row. Promoted is an
|
||||
// existing candidate this capture turned into open work: he stated out loud a
|
||||
// task Maven had only proposed, which is a confirmation, and the caller says so
|
||||
// instead of "уже в списке".
|
||||
type CaptureResult struct {
|
||||
ID int64
|
||||
Created bool
|
||||
Promoted bool
|
||||
}
|
||||
|
||||
var (
|
||||
@@ -69,19 +87,49 @@ var (
|
||||
// liveTaskStatuses — the two statuses that count as outstanding work.
|
||||
var liveTaskStatuses = []string{TaskCandidate, TaskOpen}
|
||||
|
||||
// CaptureTask inserts a task, or returns the existing live task when the same
|
||||
// work is already outstanding. created reports which happened, so a caller can
|
||||
// tell the owner "уже в списке" instead of pretending it wrote something.
|
||||
// derivedSourcePrefixes — provenance that means "Maven read this somewhere",
|
||||
// as opposed to "he said it". A task from one of these is a candidate and
|
||||
// nothing else; see CaptureTask.
|
||||
var derivedSourcePrefixes = []string{"email:"}
|
||||
|
||||
// IsDerivedSource reports whether a task source means Maven inferred the task
|
||||
// from something she read rather than being told it.
|
||||
func IsDerivedSource(source string) bool {
|
||||
for _, p := range derivedSourcePrefixes {
|
||||
if strings.HasPrefix(source, p) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// CaptureTask inserts a task, or returns the existing one when the same work is
|
||||
// already there. The result says which happened, so a caller can tell the owner
|
||||
// "уже в списке" instead of pretending it wrote something.
|
||||
//
|
||||
// Dedupe is on the normalised text among LIVE rows only (see the partial unique
|
||||
// index in migration #14): a weekly errand can be captured again once the last
|
||||
// one is done, but a mail that gets re-read produces no second row. This is the
|
||||
// property the email intake depends on — it may call CaptureTask for every
|
||||
// message it extracts from, as often as it likes, without growing the list.
|
||||
func (s *Store) CaptureTask(ctx context.Context, t Task) (id int64, created bool, err error) {
|
||||
// Two dedupe keys, because voice and mail have different intake semantics:
|
||||
//
|
||||
// - ExternalID, unique over EVERY row whatever its status. A source that
|
||||
// re-reads the same immutable text forever must never resurrect work he has
|
||||
// already finished. This is the property the email intake depends on: it
|
||||
// may call CaptureTask for every message it extracts from, as often as it
|
||||
// likes, without growing the list.
|
||||
// - The normalised text among LIVE rows only (the partial unique index in
|
||||
// migration #14), for anything with no external identity. A weekly errand
|
||||
// captured again once the last one is done must produce a new row, because
|
||||
// him saying it again IS the recurrence signal.
|
||||
//
|
||||
// A capture with Status open over an existing candidate PROMOTES it. Stating
|
||||
// the work out loud is a confirmation, and leaving it a candidate would have
|
||||
// Maven read it back as something he never confirmed.
|
||||
//
|
||||
// A derived source may only ever capture a candidate. The doc on the intake
|
||||
// seam said "must NOT set Status open"; this is where that stops being an
|
||||
// honour system, so a compromised reader cannot file work he never reviewed.
|
||||
func (s *Store) CaptureTask(ctx context.Context, t Task) (CaptureResult, error) {
|
||||
text := strings.TrimSpace(t.Text)
|
||||
if text == "" {
|
||||
return 0, false, ErrTaskEmpty
|
||||
return CaptureResult{}, ErrTaskEmpty
|
||||
}
|
||||
status := t.Status
|
||||
if status == "" {
|
||||
@@ -90,7 +138,10 @@ func (s *Store) CaptureTask(ctx context.Context, t Task) (id int64, created bool
|
||||
if status != TaskCandidate && status != TaskOpen {
|
||||
// Capturing straight into a resolved state is meaningless — a task is
|
||||
// captured live and moved later.
|
||||
return 0, false, fmt.Errorf("%w: capture status %q", ErrTaskStatus, status)
|
||||
return CaptureResult{}, fmt.Errorf("%w: capture status %q", ErrTaskStatus, status)
|
||||
}
|
||||
if status == TaskOpen && IsDerivedSource(t.Source) {
|
||||
return CaptureResult{}, fmt.Errorf("%w: derived source %q may only capture a candidate", ErrTaskStatus, t.Source)
|
||||
}
|
||||
norm := NormalizeTaskText(text)
|
||||
created2 := t.CreatedTs
|
||||
@@ -101,31 +152,69 @@ func (s *Store) CaptureTask(ctx context.Context, t Task) (id int64, created bool
|
||||
if t.Due != nil {
|
||||
due = sql.NullInt64{Int64: t.Due.UnixMilli(), Valid: true}
|
||||
}
|
||||
var ext sql.NullString
|
||||
if e := strings.TrimSpace(t.ExternalID); e != "" {
|
||||
ext = sql.NullString{String: e, Valid: true}
|
||||
}
|
||||
|
||||
// Untargeted DO NOTHING: either unique index may be the one that fires, and
|
||||
// the lookup below sorts out which.
|
||||
res, err := s.db.ExecContext(ctx,
|
||||
`INSERT INTO tasks (created_ts, text, norm, source, evidence, status, due_ts, weight)
|
||||
VALUES (?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT (norm) WHERE status IN ('candidate','open') DO NOTHING`,
|
||||
created2.UnixMilli(), text, norm, t.Source, t.Evidence, status, due, t.Weight)
|
||||
`INSERT INTO tasks (created_ts, text, norm, source, evidence, ext_id, status, due_ts, weight)
|
||||
VALUES (?,?,?,?,?,?,?,?,?)
|
||||
ON CONFLICT DO NOTHING`,
|
||||
created2.UnixMilli(), text, norm, t.Source, t.Evidence, ext, status, due, t.Weight)
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("capture task: %w", err)
|
||||
return CaptureResult{}, fmt.Errorf("capture task: %w", err)
|
||||
}
|
||||
if n, err := res.RowsAffected(); err != nil {
|
||||
return 0, false, fmt.Errorf("capture task: rows affected: %w", err)
|
||||
return CaptureResult{}, fmt.Errorf("capture task: rows affected: %w", err)
|
||||
} else if n > 0 {
|
||||
id, err := res.LastInsertId()
|
||||
if err != nil {
|
||||
return 0, false, fmt.Errorf("capture task: last insert id: %w", err)
|
||||
return CaptureResult{}, fmt.Errorf("capture task: last insert id: %w", err)
|
||||
}
|
||||
return id, true, nil
|
||||
return CaptureResult{ID: id, Created: true}, nil
|
||||
}
|
||||
|
||||
// Already live — hand back the row that won.
|
||||
existing, err := s.lookupLiveTaskByNorm(ctx, norm)
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
// Something already holds one of the keys. External identity first: that
|
||||
// row may be resolved, in which case the answer is "already handled", not a
|
||||
// new task.
|
||||
var existing Task
|
||||
if ext.Valid {
|
||||
existing, err = s.lookupTaskByExternalID(ctx, ext.String)
|
||||
if err != nil && !errors.Is(err, ErrTaskNotFound) {
|
||||
return CaptureResult{}, err
|
||||
}
|
||||
}
|
||||
return existing.ID, false, nil
|
||||
if existing.ID == 0 {
|
||||
existing, err = s.lookupLiveTaskByNorm(ctx, norm)
|
||||
if err != nil {
|
||||
return CaptureResult{}, err
|
||||
}
|
||||
}
|
||||
if status == TaskOpen && existing.Status == TaskCandidate {
|
||||
if err := s.SetTaskStatus(ctx, existing.ID, TaskOpen, created2, t.Source); err != nil {
|
||||
return CaptureResult{}, fmt.Errorf("capture task: promote candidate: %w", err)
|
||||
}
|
||||
return CaptureResult{ID: existing.ID, Promoted: true}, nil
|
||||
}
|
||||
return CaptureResult{ID: existing.ID}, nil
|
||||
}
|
||||
|
||||
// lookupTaskByExternalID finds a task by the identity of what it was derived
|
||||
// from, in ANY status. Resolved rows count: the whole point of the key is that
|
||||
// re-reading the mail that produced a finished task produces nothing.
|
||||
func (s *Store) lookupTaskByExternalID(ctx context.Context, ext string) (Task, error) {
|
||||
row := s.db.QueryRowContext(ctx, taskSelect+` WHERE ext_id = ?`, ext)
|
||||
t, err := scanTask(row)
|
||||
if errors.Is(err, sql.ErrNoRows) {
|
||||
return Task{}, ErrTaskNotFound
|
||||
}
|
||||
if err != nil {
|
||||
return Task{}, fmt.Errorf("lookup task by external id: %w", err)
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// lookupLiveTaskByNorm finds the outstanding task with this normalised text.
|
||||
@@ -142,7 +231,7 @@ func (s *Store) lookupLiveTaskByNorm(ctx context.Context, norm string) (Task, er
|
||||
return t, nil
|
||||
}
|
||||
|
||||
const taskSelect = `SELECT id, created_ts, text, source, evidence, status, due_ts, weight, resolved_ts FROM tasks`
|
||||
const taskSelect = `SELECT id, created_ts, text, source, evidence, COALESCE(ext_id,''), status, due_ts, weight, resolved_ts, resolved_by FROM tasks`
|
||||
|
||||
// LookupTask returns one task by id.
|
||||
func (s *Store) LookupTask(ctx context.Context, id int64) (Task, error) {
|
||||
@@ -157,9 +246,15 @@ func (s *Store) LookupTask(ctx context.Context, id int64) (Task, error) {
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// ListTasks returns tasks in one status, newest first. An empty status returns
|
||||
// every row; "live" returns candidate + open, which is what every read path
|
||||
// that means "outstanding work" wants.
|
||||
// MaxTaskRows — the hard bound on one ListTasks read. The live set is a list a
|
||||
// person keeps by hand and never approaches this; the resolved set grows for as
|
||||
// long as the box runs, and an unbounded read of it is a page that gets slower
|
||||
// every month. Newest first, so the bound drops the oldest finished work.
|
||||
const MaxTaskRows = 500
|
||||
|
||||
// ListTasks returns tasks in one status, newest first, at most MaxTaskRows of
|
||||
// them. An empty status returns every row; "live" returns candidate + open,
|
||||
// which is what every read path that means "outstanding work" wants.
|
||||
func (s *Store) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
||||
q := taskSelect
|
||||
var args []any
|
||||
@@ -172,7 +267,7 @@ func (s *Store) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
||||
q += ` WHERE status = ?`
|
||||
args = append(args, status)
|
||||
}
|
||||
q += ` ORDER BY created_ts DESC, id DESC`
|
||||
q += fmt.Sprintf(` ORDER BY created_ts DESC, id DESC LIMIT %d`, MaxTaskRows)
|
||||
|
||||
rows, err := s.db.QueryContext(ctx, q, args...)
|
||||
if err != nil {
|
||||
@@ -201,8 +296,14 @@ func (s *Store) ListTasks(ctx context.Context, status string) ([]Task, error) {
|
||||
// ErrTaskNotFound-wrapped detail, the same one-way shape proposed_routines and
|
||||
// tools use: an answered question is not answered twice.
|
||||
//
|
||||
// Resolving frees the dedupe key, which is the point: the work can recur.
|
||||
func (s *Store) SetTaskStatus(ctx context.Context, id int64, status string, ts time.Time) error {
|
||||
// Resolving frees the NORM dedupe key, which is the point: the work can recur
|
||||
// when he says it again. It does not free an external identity — see
|
||||
// CaptureTask for why a re-read mailbox must not resurrect finished work.
|
||||
//
|
||||
// by names the caller making the move, in the source vocabulary ("tap:web",
|
||||
// "tap:voice"). It is recorded on the row, so a task that turns up resolved
|
||||
// says what resolved it.
|
||||
func (s *Store) SetTaskStatus(ctx context.Context, id int64, status string, ts time.Time, by string) error {
|
||||
var from []string
|
||||
switch status {
|
||||
case TaskOpen:
|
||||
@@ -222,9 +323,9 @@ func (s *Store) SetTaskStatus(ctx context.Context, id int64, status string, ts t
|
||||
resolved = sql.NullInt64{Int64: ts.UnixMilli(), Valid: true}
|
||||
}
|
||||
|
||||
q := `UPDATE tasks SET status = ?, resolved_ts = ? WHERE id = ? AND status IN (?` +
|
||||
q := `UPDATE tasks SET status = ?, resolved_ts = ?, resolved_by = ? WHERE id = ? AND status IN (?` +
|
||||
strings.Repeat(",?", len(from)-1) + `)`
|
||||
args := []any{status, resolved, id}
|
||||
args := []any{status, resolved, by, id}
|
||||
for _, f := range from {
|
||||
args = append(args, f)
|
||||
}
|
||||
@@ -271,7 +372,7 @@ func scanTask(sc scanner) (Task, error) {
|
||||
var t Task
|
||||
var created int64
|
||||
var due, resolved sql.NullInt64
|
||||
if err := sc.Scan(&t.ID, &created, &t.Text, &t.Source, &t.Evidence, &t.Status, &due, &t.Weight, &resolved); err != nil {
|
||||
if err := sc.Scan(&t.ID, &created, &t.Text, &t.Source, &t.Evidence, &t.ExternalID, &t.Status, &due, &t.Weight, &resolved, &t.ResolvedBy); err != nil {
|
||||
return Task{}, err
|
||||
}
|
||||
t.CreatedTs = time.UnixMilli(created).UTC()
|
||||
|
||||
Reference in New Issue
Block a user