62 lines
1.6 KiB
Go
62 lines
1.6 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"g.fc.run/ai/byre/internal/api"
|
|
)
|
|
|
|
func (s *Store) SummaryReads(ctx context.Context, login string) (map[string]int64, error) {
|
|
rows, err := s.db.QueryContext(ctx, `
|
|
SELECT summary_reads.task_id, summary_reads.seq FROM summary_reads
|
|
JOIN tasks ON tasks.id = summary_reads.task_id
|
|
WHERE summary_reads.login = ? AND tasks.status = ?
|
|
`, login, string(api.TaskStatusActive))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
reads := map[string]int64{}
|
|
for rows.Next() {
|
|
taskID := ""
|
|
seq := int64(0)
|
|
err = rows.Scan(&taskID, &seq)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
reads[taskID] = seq
|
|
}
|
|
return reads, rows.Err()
|
|
}
|
|
|
|
func (s *Store) MarkSummaryRead(ctx context.Context, login, taskID string, seq int64) error {
|
|
return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
|
|
err := tx.QueryRowContext(ctx, `SELECT id FROM tasks WHERE id = ?`, taskID).Scan(&taskID)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("task %s: %w", taskID, ErrNotFound)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
res, err := tx.ExecContext(ctx, `
|
|
INSERT INTO summary_reads (login, task_id, seq) VALUES (?, ?, ?)
|
|
ON CONFLICT (login, task_id) DO UPDATE SET seq = excluded.seq WHERE excluded.seq > summary_reads.seq
|
|
`, login, taskID, seq)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
n, err := res.RowsAffected()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if n == 0 {
|
|
return nil
|
|
}
|
|
return emit(api.Event{Kind: "summary.read", Owner: &login, Actor: login, Payload: map[string]any{"task_id": taskID, "seq": seq}})
|
|
})
|
|
}
|