package store import ( "context" "database/sql" "encoding/json" "errors" "time" "g.fc.run/ai/byre/internal/api" ) type LeadInputKind string const ( LeadMessage LeadInputKind = "message" LeadSignal LeadInputKind = "signal" ) type LeadInput struct { ID int64 Owner string Kind LeadInputKind Actor string Text string Images []string SendID string Client string CreatedAt time.Time } func (s *Store) AddLeadInput(ctx context.Context, in LeadInput) error { images, err := json.Marshal(append([]string{}, in.Images...)) if err != nil { 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, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, in.Owner, string(in.Kind), in.Actor, in.Text, string(images), in.SendID, in.Client, 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, send_id, client_id, created_at FROM lead_inputs WHERE owner = ? ORDER BY id`, owner) if err != nil { return nil, err } defer rows.Close() inputs := []LeadInput{} for rows.Next() { in := LeadInput{} kind := "" images := "" createdAt := "" err = rows.Scan(&in.ID, &in.Owner, &kind, &in.Actor, &in.Text, &images, &in.SendID, &in.Client, &createdAt) if err != nil { return nil, err } in.Kind = LeadInputKind(kind) err = json.Unmarshal([]byte(images), &in.Images) if err != nil { return nil, err } in.CreatedAt, err = parseTime(createdAt) if err != nil { return nil, err } inputs = append(inputs, in) } return inputs, rows.Err() } func (s *Store) RestoreLeadInputs(ctx context.Context, inputs []LeadInput) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { for _, in := range inputs { images, err := json.Marshal(append([]string{}, in.Images...)) if err != nil { return err } _, err = tx.ExecContext(ctx, `INSERT INTO lead_inputs (id, owner, kind, actor, text, images, send_id, client_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, in.ID, in.Owner, string(in.Kind), in.Actor, in.Text, string(images), in.SendID, in.Client, formatTime(in.CreatedAt)) if err != nil { return err } } return nil }) } type LeadDelivery struct { Owner string First int64 Last int64 } func leadDeliveries(ctx context.Context, tx *sql.Tx) ([]LeadDelivery, error) { rows, err := tx.QueryContext(ctx, ` SELECT i.owner, min(i.id), max(i.id) FROM lead_inputs i WHERE (SELECT l.state FROM agent_sessions l WHERE l.owner = i.owner AND l.state != 'completed' ORDER BY l.created_at DESC LIMIT 1) = 'idle' GROUP BY i.owner`) if err != nil { return nil, err } defer rows.Close() found := []LeadDelivery{} for rows.Next() { d := LeadDelivery{} err = rows.Scan(&d.Owner, &d.First, &d.Last) if err != nil { return nil, err } found = append(found, d) } return found, rows.Err() } func (s *Store) DeleteLeadInputs(ctx context.Context, ids []int64) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { for _, id := range ids { _, err := tx.ExecContext(ctx, `DELETE FROM lead_inputs WHERE id = ?`, id) if err != nil { return err } } return nil }) } func (s *Store) AddLeadWake(ctx context.Context, owner, taskID string) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { active := false err := tx.QueryRowContext(ctx, `SELECT status = ? FROM tasks WHERE id = ?`, string(api.TaskStatusActive), taskID).Scan(&active) if errors.Is(err, sql.ErrNoRows) || (err == nil && !active) { return ErrNotFound } if err != nil { return err } _, err = tx.ExecContext(ctx, `INSERT OR IGNORE INTO lead_wakes (owner, task_id, created_at) VALUES (?, ?, ?)`, owner, taskID, now()) return err }) } func (s *Store) TakeLeadWakes(ctx context.Context, taskID string) ([]string, error) { owners := []string{} err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { rows, err := tx.QueryContext(ctx, `DELETE FROM lead_wakes WHERE task_id = ? RETURNING owner`, taskID) if err != nil { return err } defer rows.Close() for rows.Next() { owner := "" err = rows.Scan(&owner) if err != nil { return err } owners = append(owners, owner) } return rows.Err() }) return owners, err } func (s *Store) GetFocus(ctx context.Context, login string) (api.Focus, error) { f := api.Focus{} var task, project sql.NullString err := s.db.QueryRowContext(ctx, `SELECT focus_task_id, focus_project_id FROM users WHERE login = ?`, login).Scan(&task, &project) if errors.Is(err, sql.ErrNoRows) { return f, ErrNotFound } if err != nil { return f, err } if task.Valid { f.TaskId = &task.String } if project.Valid { f.ProjectId = &project.String } return f, nil } func (s *Store) SetFocus(ctx context.Context, login string, f api.Focus, actor string) error { return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error { _, err := tx.ExecContext(ctx, `UPDATE users SET focus_task_id = ?, focus_project_id = ? WHERE login = ?`, f.TaskId, f.ProjectId, login) if err != nil { return err } return emit(api.Event{Kind: "focus.updated", Owner: &login, Actor: actor, Payload: ToPayload(f)}) }) }