ai/byre
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Match pending sends by a send id the server echoes

Browse Source
This commit is contained in:
Ian Gulliver
2026-10-20 15:22:50 -07:00
parent b8c5331a0c
commit 8fec4b9040
29 changed files with 437 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/store/lead.go
+7 -6
View File
@@ -24,6 +24,7 @@ type LeadInput struct {
Actor string
Text string
Images []string
SendID string
CreatedAt time.Time
}
@@ -33,14 +34,14 @@ func (s *Store) AddLeadInput(ctx context.Context, in LeadInput) error {
return err
}
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
_, err := tx.ExecContext(ctx, `INSERT INTO lead_inputs (owner, kind, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?)`,
in.Owner, string(in.Kind), in.Actor, in.Text, string(images), now())
_, err := tx.ExecContext(ctx, `INSERT INTO lead_inputs (owner, kind, actor, text, images, send_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
in.Owner, string(in.Kind), in.Actor, in.Text, string(images), in.SendID, now())
return err
})
}
func (s *Store) LeadInputs(ctx context.Context, owner string) ([]LeadInput, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, owner, kind, actor, text, images, created_at FROM lead_inputs WHERE owner = ? ORDER BY id`, owner)
rows, err := s.db.QueryContext(ctx, `SELECT id, owner, kind, actor, text, images, send_id, created_at FROM lead_inputs WHERE owner = ? ORDER BY id`, owner)
if err != nil {
return nil, err
}
@@ -52,7 +53,7 @@ func (s *Store) LeadInputs(ctx context.Context, owner string) ([]LeadInput, erro
kind := ""
images := ""
createdAt := ""
err = rows.Scan(&in.ID, &in.Owner, &kind, &in.Actor, &in.Text, &images, &createdAt)
err = rows.Scan(&in.ID, &in.Owner, &kind, &in.Actor, &in.Text, &images, &in.SendID, &createdAt)
if err != nil {
return nil, err
}
@@ -77,8 +78,8 @@ func (s *Store) RestoreLeadInputs(ctx context.Context, inputs []LeadInput) error
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO lead_inputs (id, owner, kind, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
in.ID, in.Owner, string(in.Kind), in.Actor, in.Text, string(images), formatTime(in.CreatedAt))
_, err = tx.ExecContext(ctx, `INSERT INTO lead_inputs (id, owner, kind, actor, text, images, send_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
in.ID, in.Owner, string(in.Kind), in.Actor, in.Text, string(images), in.SendID, formatTime(in.CreatedAt))
if err != nil {
return err
}
internal/store/migrations/069-add-input-send-ids.sql
+33
View File
@@ -0,0 +1,33 @@
ALTER TABLE task_inputs ADD COLUMN send_id TEXT NOT NULL DEFAULT '';
ALTER TABLE lead_inputs ADD COLUMN send_id TEXT NOT NULL DEFAULT '';
UPDATE task_inputs SET send_id = 'queued-' || id;
UPDATE events SET payload = json_set(payload, '$.send_id', 'queued-' || json_extract(payload, '$.id'))
WHERE kind = 'user.queued';
-- The n-th delivered queue entry of a task, sender and text pairs with the n-th later user.message of the same.
WITH q AS (
SELECT e.seq, e.task_id, e.actor, json_extract(e.payload, '$.text') AS text, json_extract(e.payload, '$.send_id') AS send_id,
row_number() OVER (PARTITION BY e.task_id, e.actor, json_extract(e.payload, '$.text') ORDER BY e.seq) AS n
FROM events AS e
WHERE e.kind = 'user.queued'
AND json_extract(e.payload, '$.id') NOT IN (SELECT id FROM task_inputs)
AND NOT EXISTS (
SELECT 1 FROM events AS u
WHERE u.kind = 'user.unqueued' AND u.task_id = e.task_id AND json_extract(u.payload, '$.id') = json_extract(e.payload, '$.id')
)
), first AS (
SELECT task_id, actor, text, min(seq) AS seq FROM q GROUP BY task_id, actor, text
), m AS (
SELECT e.seq, e.task_id, e.actor, json_extract(e.payload, '$.text') AS text,
row_number() OVER (PARTITION BY e.task_id, e.actor, json_extract(e.payload, '$.text') ORDER BY e.seq) AS n
FROM events AS e
JOIN first AS f ON f.task_id = e.task_id AND f.actor = e.actor AND f.text = json_extract(e.payload, '$.text') AND e.seq > f.seq
WHERE e.kind = 'user.message'
), pairs AS (
SELECT m.seq, q.send_id FROM m
JOIN q ON q.task_id = m.task_id AND q.actor = m.actor AND q.text = m.text AND q.n = m.n AND q.seq < m.seq
)
UPDATE events SET payload = json_set(payload, '$.send_id', (SELECT send_id FROM pairs WHERE pairs.seq = events.seq))
WHERE seq IN (SELECT seq FROM pairs);
internal/store/taskinputs.go
+11 -7
View File
@@ -15,6 +15,7 @@ type TaskInput struct {
Actor string
Text string
Images []string
SendID string
CreatedAt time.Time
}
@@ -23,13 +24,16 @@ func (s *Store) AddTaskInput(ctx context.Context, in TaskInput, projectID, sessi
if err != nil {
return err
}
payload := map[string]any{"text": in.Text}
if in.SendID == "" {
in.SendID = NewID()
}
payload := map[string]any{"text": in.Text, "send_id": in.SendID}
if len(in.Images) > 0 {
payload["images"] = in.Images
}
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
res, err := tx.ExecContext(ctx, `INSERT INTO task_inputs (task_id, member, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?)`,
in.TaskID, in.Member, in.Actor, in.Text, string(images), now())
res, err := tx.ExecContext(ctx, `INSERT INTO task_inputs (task_id, member, actor, text, images, send_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
in.TaskID, in.Member, in.Actor, in.Text, string(images), in.SendID, now())
if err != nil {
return err
}
@@ -62,7 +66,7 @@ func (s *Store) WithdrawTaskInput(ctx context.Context, taskID string, id int64,
}
func (s *Store) TaskInputs(ctx context.Context, taskID, member string) ([]TaskInput, error) {
rows, err := s.db.QueryContext(ctx, `SELECT id, task_id, member, actor, text, images, created_at FROM task_inputs WHERE task_id = ? AND member = ? ORDER BY id`, taskID, member)
rows, err := s.db.QueryContext(ctx, `SELECT id, task_id, member, actor, text, images, send_id, created_at FROM task_inputs WHERE task_id = ? AND member = ? ORDER BY id`, taskID, member)
if err != nil {
return nil, err
}
@@ -73,7 +77,7 @@ func (s *Store) TaskInputs(ctx context.Context, taskID, member string) ([]TaskIn
in := TaskInput{}
images := ""
createdAt := ""
err = rows.Scan(&in.ID, &in.TaskID, &in.Member, &in.Actor, &in.Text, &images, &createdAt)
err = rows.Scan(&in.ID, &in.TaskID, &in.Member, &in.Actor, &in.Text, &images, &in.SendID, &createdAt)
if err != nil {
return nil, err
}
@@ -97,8 +101,8 @@ func (s *Store) RestoreTaskInputs(ctx context.Context, inputs []TaskInput) error
if err != nil {
return err
}
_, err = tx.ExecContext(ctx, `INSERT INTO task_inputs (id, task_id, member, actor, text, images, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
in.ID, in.TaskID, in.Member, in.Actor, in.Text, string(images), formatTime(in.CreatedAt))
_, err = tx.ExecContext(ctx, `INSERT INTO task_inputs (id, task_id, member, actor, text, images, send_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
in.ID, in.TaskID, in.Member, in.Actor, in.Text, string(images), in.SendID, formatTime(in.CreatedAt))
if err != nil {
return err
}