byre/internal/store/taskinputs.go

package store

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"time"
)

type TaskInput struct {
	ID        int64
	TaskID    string
	Member    string
	Actor     string
	Text      string
	Images    []string
	SendID    string
	CreatedAt time.Time
}

func (s *Store) AddTaskInput(ctx context.Context, in TaskInput, projectID, sessionID string) error {
	images, err := json.Marshal(append([]string{}, in.Images...))
	if err != nil {
		return err
	}
	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, send_id, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)`,
			in.TaskID, in.Member, in.Actor, in.Text, string(images), in.SendID, now())
		if err != nil {
			return err
		}
		payload["id"], err = res.LastInsertId()
		if err != nil {
			return err
		}
		return emit(sessionEvent("user.queued", projectID, in.TaskID, "", sessionID, in.Actor, payload))
	})
}

func (s *Store) SendSession(ctx context.Context, sendID string) (string, bool, error) {
	sessionID := sql.NullString{}
	err := s.db.QueryRowContext(ctx, `SELECT agent_session_id FROM events WHERE kind IN ('user.queued', 'user.message') AND json_extract(payload, '$.send_id') = ? LIMIT 1`, sendID).Scan(&sessionID)
	if err == nil {
		return sessionID.String, true, nil
	}
	if !errors.Is(err, sql.ErrNoRows) {
		return "", false, err
	}
	err = s.db.QueryRowContext(ctx, `SELECT 1 FROM lead_inputs WHERE send_id = ? LIMIT 1`, sendID).Scan(new(int))
	if errors.Is(err, sql.ErrNoRows) {
		return "", false, nil
	}
	return "", err == nil, err
}

func (s *Store) WithdrawTaskInput(ctx context.Context, taskID string, id int64, projectID, actor string) (string, error) {
	sessionID := ""
	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
		member := ""
		err := tx.QueryRowContext(ctx, `DELETE FROM task_inputs WHERE id = ? AND task_id = ? RETURNING member`, id, taskID).Scan(&member)
		if errors.Is(err, sql.ErrNoRows) {
			return ErrNotFound
		}
		if err != nil {
			return err
		}
		err = tx.QueryRowContext(ctx, `SELECT id FROM agent_sessions WHERE task_id = ? AND member = ? AND role = 'implementer' ORDER BY created_at DESC LIMIT 1`, taskID, member).Scan(&sessionID)
		if err != nil {
			return err
		}
		return emit(sessionEvent("user.unqueued", projectID, taskID, "", sessionID, actor, map[string]any{"id": id}))
	})
	return sessionID, err
}

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, send_id, created_at FROM task_inputs WHERE task_id = ? AND member = ? ORDER BY id`, taskID, member)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	inputs := []TaskInput{}
	for rows.Next() {
		in := TaskInput{}
		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
		}
		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) RestoreTaskInputs(ctx context.Context, inputs []TaskInput) 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 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
			}
		}
		return nil
	})
}

func (s *Store) DeleteTaskInputs(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 task_inputs WHERE id = ?`, id)
			if err != nil {
				return err
			}
		}
		return nil
	})
}