package store import ( "context" "database/sql" "encoding/json" "errors" "fmt" "time" "gitea.stevedudenhoeffer.com/steve/executus/skill" ) // skillStore is the SQLite-backed skill.SkillStore. Same JSON-blob + indexed // columns approach as personaStore: the full Skill round-trips, lookups stay // indexed. Versions live in their own table (each SkillVersion embeds a full // Skill snapshot, stored as a JSON blob). type skillStore struct{ db *sql.DB } // Skills returns a durable skill.SkillStore backed by this database. func (d *DB) Skills() skill.SkillStore { return &skillStore{db: d.sql} } var _ skill.SkillStore = (*skillStore)(nil) func (s *skillStore) Initialize(ctx context.Context) error { _, err := s.db.ExecContext(ctx, ` CREATE TABLE IF NOT EXISTS skills ( id TEXT PRIMARY KEY, owner_id TEXT NOT NULL DEFAULT '', name TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT '', visibility TEXT NOT NULL DEFAULT '', chatbot INTEGER NOT NULL DEFAULT 0, -- ExposeAsChatbotTool schedule TEXT NOT NULL DEFAULT '', next_run_at INTEGER NOT NULL DEFAULT 0, data TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_skills_owner ON skills(owner_id); CREATE INDEX IF NOT EXISTS idx_skills_vis ON skills(visibility); CREATE INDEX IF NOT EXISTS idx_skills_sched ON skills(schedule, next_run_at); CREATE TABLE IF NOT EXISTS skill_versions ( id TEXT PRIMARY KEY, skill_id TEXT NOT NULL, version TEXT NOT NULL DEFAULT '', seq INTEGER NOT NULL, -- append order, for newest-first data TEXT NOT NULL ); CREATE UNIQUE INDEX IF NOT EXISTS idx_skill_versions_skill ON skill_versions(skill_id, seq);`) if err != nil { return fmt.Errorf("skillStore.Initialize: %w", err) } return nil } func (s *skillStore) Save(ctx context.Context, sk *skill.Skill) error { blob, err := json.Marshal(sk) if err != nil { return fmt.Errorf("skillStore.Save: marshal: %w", err) } var next int64 if !sk.NextRunAt.IsZero() { next = sk.NextRunAt.Unix() } chatbot := 0 if sk.ExposeAsChatbotTool { chatbot = 1 } _, err = s.db.ExecContext(ctx, ` INSERT INTO skills (id, owner_id, name, source, visibility, chatbot, schedule, next_run_at, data) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET owner_id=excluded.owner_id, name=excluded.name, source=excluded.source, visibility=excluded.visibility, chatbot=excluded.chatbot, schedule=excluded.schedule, next_run_at=excluded.next_run_at, data=excluded.data`, sk.ID, sk.OwnerID, sk.Name, string(sk.Source), string(sk.Visibility), chatbot, sk.Schedule, next, string(blob)) if err != nil { return fmt.Errorf("skillStore.Save: %w", err) } return nil } func scanSkills(rows *sql.Rows) ([]skill.Skill, error) { defer rows.Close() var out []skill.Skill for rows.Next() { var blob string if err := rows.Scan(&blob); err != nil { return nil, err } var sk skill.Skill if err := json.Unmarshal([]byte(blob), &sk); err != nil { return nil, err } out = append(out, sk) } return out, rows.Err() } func (s *skillStore) getOne(ctx context.Context, where string, arg ...any) (*skill.Skill, error) { var blob string err := s.db.QueryRowContext(ctx, `SELECT data FROM skills WHERE `+where, arg...).Scan(&blob) switch { case errors.Is(err, sql.ErrNoRows): return nil, skill.ErrNotFound case err != nil: return nil, err } var sk skill.Skill if err := json.Unmarshal([]byte(blob), &sk); err != nil { return nil, err } return &sk, nil } func (s *skillStore) Get(ctx context.Context, id string) (*skill.Skill, error) { return s.getOne(ctx, "id = ?", id) } func (s *skillStore) GetByName(ctx context.Context, ownerID, name string) (*skill.Skill, error) { return s.getOne(ctx, "owner_id = ? AND name = ?", ownerID, name) } func (s *skillStore) ListBuiltinByName(ctx context.Context, name string) (*skill.Skill, error) { return s.getOne(ctx, "source = ? AND name = ?", string(skill.SourceBuiltin), name) } func (s *skillStore) Delete(ctx context.Context, id string) error { if _, err := s.db.ExecContext(ctx, `DELETE FROM skills WHERE id = ?`, id); err != nil { return fmt.Errorf("skillStore.Delete: %w", err) } return nil } func (s *skillStore) query(ctx context.Context, where string, arg ...any) ([]skill.Skill, error) { rows, err := s.db.QueryContext(ctx, `SELECT data FROM skills WHERE `+where+` ORDER BY name`, arg...) if err != nil { return nil, err } return scanSkills(rows) } func (s *skillStore) ListByOwner(ctx context.Context, ownerID string) ([]skill.Skill, error) { return s.query(ctx, "owner_id = ?", ownerID) } func (s *skillStore) ListPublic(ctx context.Context) ([]skill.Skill, error) { return s.query(ctx, "visibility = ?", string(skill.VisibilityPublic)) } func (s *skillStore) ListChatbotExposed(ctx context.Context) ([]skill.Skill, error) { return s.query(ctx, "chatbot = 1") } // ListSharedWith loads visibility=shared rows and filters SharedWith in Go (the // shared set per skill is small; avoids a JSON-array query). func (s *skillStore) ListSharedWith(ctx context.Context, memberID string) ([]skill.Skill, error) { shared, err := s.query(ctx, "visibility = ?", string(skill.VisibilityShared)) if err != nil { return nil, err } out := shared[:0] for _, sk := range shared { for _, id := range sk.SharedWith { if id == memberID { out = append(out, sk) break } } } return out, nil } func (s *skillStore) ListDueScheduled(ctx context.Context, now time.Time) ([]skill.Skill, error) { rows, err := s.db.QueryContext(ctx, `SELECT data FROM skills WHERE schedule != '' AND next_run_at > 0 AND next_run_at <= ? ORDER BY next_run_at`, now.Unix()) if err != nil { return nil, fmt.Errorf("skillStore.ListDueScheduled: %w", err) } return scanSkills(rows) } func (s *skillStore) MarkScheduledRun(ctx context.Context, skillID string, ranAt, nextAt time.Time) error { // Single atomic statement instead of Get→mutate→Save: a concurrent Mark or // admin edit can't lose this update (no read-modify-write window). json_set // keeps the JSON blob's NextRunAt/LastScheduledRunAt consistent with the // indexed next_run_at column; RFC3339Nano matches Go's time JSON encoding so // the blob still round-trips through Get. var next int64 if !nextAt.IsZero() { next = nextAt.Unix() } res, err := s.db.ExecContext(ctx, `UPDATE skills SET next_run_at=?, data=json_set(data,'$.NextRunAt',?,'$.LastScheduledRunAt',?) WHERE id=?`, next, nextAt.Format(time.RFC3339Nano), ranAt.Format(time.RFC3339Nano), skillID) if err != nil { return fmt.Errorf("skillStore.MarkScheduledRun: %w", err) } if n, _ := res.RowsAffected(); n == 0 { return skill.ErrNotFound } return nil } func (s *skillStore) AppendVersion(ctx context.Context, sv skill.SkillVersion) error { if sv.SkillID == "" { return fmt.Errorf("skillStore.AppendVersion: skill_id is required") } blob, err := json.Marshal(sv) if err != nil { return fmt.Errorf("skillStore.AppendVersion: marshal: %w", err) } // seq = current max+1 for this skill (newest-first ordering key). The // MAX-then-INSERT runs in ONE transaction and the (skill_id, seq) index is // UNIQUE, so two concurrent appends can't both land the same seq: the loser // fails loudly on commit instead of silently corrupting the ordering. The // Scan error is propagated (was swallowed, leaving seq=0 on failure). tx, err := s.db.BeginTx(ctx, nil) if err != nil { return fmt.Errorf("skillStore.AppendVersion: begin: %w", err) } defer tx.Rollback() //nolint:errcheck // no-op after Commit var seq int64 if err := tx.QueryRowContext(ctx, `SELECT COALESCE(MAX(seq),0)+1 FROM skill_versions WHERE skill_id = ?`, sv.SkillID).Scan(&seq); err != nil { return fmt.Errorf("skillStore.AppendVersion: seq: %w", err) } if _, err := tx.ExecContext(ctx, `INSERT INTO skill_versions (id, skill_id, version, seq, data) VALUES (?, ?, ?, ?, ?)`, sv.ID, sv.SkillID, sv.Version, seq, string(blob)); err != nil { return fmt.Errorf("skillStore.AppendVersion: insert: %w", err) } if err := tx.Commit(); err != nil { return fmt.Errorf("skillStore.AppendVersion: commit: %w", err) } return nil } func (s *skillStore) ListVersionsBySkill(ctx context.Context, skillID string, limit int) ([]skill.SkillVersion, error) { q := `SELECT data FROM skill_versions WHERE skill_id = ? ORDER BY seq DESC` args := []any{skillID} if limit > 0 { q += ` LIMIT ?` args = append(args, limit) } rows, err := s.db.QueryContext(ctx, q, args...) if err != nil { return nil, fmt.Errorf("skillStore.ListVersionsBySkill: %w", err) } defer rows.Close() var out []skill.SkillVersion for rows.Next() { var blob string if err := rows.Scan(&blob); err != nil { return nil, err } var sv skill.SkillVersion if err := json.Unmarshal([]byte(blob), &sv); err != nil { return nil, err } out = append(out, sv) } return out, rows.Err() } func (s *skillStore) GetVersionByID(ctx context.Context, versionID string) (*skill.SkillVersion, error) { var blob string err := s.db.QueryRowContext(ctx, `SELECT data FROM skill_versions WHERE id = ?`, versionID).Scan(&blob) switch { case errors.Is(err, sql.ErrNoRows): return nil, skill.ErrNotFound case err != nil: return nil, err } var sv skill.SkillVersion if err := json.Unmarshal([]byte(blob), &sv); err != nil { return nil, err } return &sv, nil }