package sqlite import ( "context" "database/sql" "encoding/json" "fmt" "time" "github.com/google/uuid" "armature/database" "armature/models" "armature/store" ) type TriggerStore struct{} func NewTriggerStore() *TriggerStore { return &TriggerStore{} } func (s *TriggerStore) Create(ctx context.Context, t *models.Trigger) error { t.ID = uuid.New().String() cfg, _ := json.Marshal(t.Config) if len(cfg) == 0 { cfg = []byte("{}") } _, err := DB.ExecContext(ctx, `INSERT INTO triggers (id, package_id, type, enabled, slug, secret, event_pattern, entry_point, config) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, t.ID, t.PackageID, t.Type, boolToInt(t.Enabled), nullIfEmpty(t.Slug), nullIfEmpty(t.Secret), nullIfEmpty(t.EventPattern), t.EntryPoint, string(cfg)) return err } func (s *TriggerStore) GetByID(ctx context.Context, id string) (*models.Trigger, error) { var t models.Trigger var cfgStr string err := DB.QueryRowContext(ctx, `SELECT id, package_id, type, enabled, COALESCE(slug,''), secret, COALESCE(event_pattern,''), entry_point, config, fire_count, last_fire_at, COALESCE(last_error,''), created_at, updated_at FROM triggers WHERE id = ?`, id). Scan(&t.ID, &t.PackageID, &t.Type, &t.Enabled, &t.Slug, &t.Secret, &t.EventPattern, &t.EntryPoint, &cfgStr, &t.FireCount, database.SNT(&t.LastFireAt), &t.LastError, database.ST(&t.CreatedAt), database.ST(&t.UpdatedAt)) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } t.Config = json.RawMessage(cfgStr) return &t, nil } func (s *TriggerStore) Update(ctx context.Context, t *models.Trigger) error { cfg, _ := json.Marshal(t.Config) if len(cfg) == 0 { cfg = []byte("{}") } _, err := DB.ExecContext(ctx, `UPDATE triggers SET enabled = ?, slug = ?, secret = ?, event_pattern = ?, entry_point = ?, config = ?, updated_at = datetime('now') WHERE id = ?`, boolToInt(t.Enabled), nullIfEmpty(t.Slug), nullIfEmpty(t.Secret), nullIfEmpty(t.EventPattern), t.EntryPoint, string(cfg), t.ID) return err } func (s *TriggerStore) Delete(ctx context.Context, id string) error { _, err := DB.ExecContext(ctx, `DELETE FROM triggers WHERE id = ?`, id) return err } func (s *TriggerStore) List(ctx context.Context, opts store.TriggerListOptions) ([]models.Trigger, int, error) { where := "1=1" args := []any{} if opts.PackageID != "" { where += " AND package_id = ?" args = append(args, opts.PackageID) } if opts.Type != "" { where += " AND type = ?" args = append(args, opts.Type) } if opts.Enabled != nil { where += " AND enabled = ?" args = append(args, boolToInt(*opts.Enabled)) } var total int countArgs := make([]any, len(args)) copy(countArgs, args) err := DB.QueryRowContext(ctx, "SELECT COUNT(*) FROM triggers WHERE "+where, countArgs...).Scan(&total) if err != nil { return nil, 0, err } limit := opts.Limit if limit <= 0 { limit = 50 } query := fmt.Sprintf( `SELECT id, package_id, type, enabled, COALESCE(slug,''), COALESCE(event_pattern,''), entry_point, config, fire_count, last_fire_at, COALESCE(last_error,''), created_at, updated_at FROM triggers WHERE %s ORDER BY created_at DESC LIMIT ? OFFSET ?`, where) args = append(args, limit, opts.Offset) rows, err := DB.QueryContext(ctx, query, args...) if err != nil { return nil, 0, err } defer rows.Close() triggers, err := scanTriggerRowsSqlite(rows) return triggers, total, err } func (s *TriggerStore) ListByPackage(ctx context.Context, packageID string) ([]models.Trigger, error) { rows, err := DB.QueryContext(ctx, `SELECT id, package_id, type, enabled, COALESCE(slug,''), COALESCE(event_pattern,''), entry_point, config, fire_count, last_fire_at, COALESCE(last_error,''), created_at, updated_at FROM triggers WHERE package_id = ? ORDER BY created_at`, packageID) if err != nil { return nil, err } defer rows.Close() return scanTriggerRowsSqlite(rows) } func (s *TriggerStore) ListEnabledByType(ctx context.Context, triggerType string) ([]models.Trigger, error) { rows, err := DB.QueryContext(ctx, `SELECT id, package_id, type, enabled, COALESCE(slug,''), COALESCE(event_pattern,''), entry_point, config, fire_count, last_fire_at, COALESCE(last_error,''), created_at, updated_at FROM triggers WHERE type = ? AND enabled = 1 ORDER BY created_at`, triggerType) if err != nil { return nil, err } defer rows.Close() return scanTriggerRowsSqlite(rows) } func (s *TriggerStore) SetEnabled(ctx context.Context, id string, enabled bool) error { _, err := DB.ExecContext(ctx, `UPDATE triggers SET enabled = ?, updated_at = datetime('now') WHERE id = ?`, boolToInt(enabled), id) return err } func (s *TriggerStore) UpdateFireState(ctx context.Context, id string, lastFireAt time.Time, lastError string) error { _, err := DB.ExecContext(ctx, `UPDATE triggers SET last_fire_at = ?, last_error = ?, fire_count = fire_count + 1, updated_at = datetime('now') WHERE id = ?`, lastFireAt.UTC().Format(time.RFC3339), nullIfEmpty(lastError), id) return err } func (s *TriggerStore) GetWebhook(ctx context.Context, packageID, slug string) (*models.Trigger, error) { var t models.Trigger var cfgStr string err := DB.QueryRowContext(ctx, `SELECT id, package_id, type, enabled, COALESCE(slug,''), secret, COALESCE(event_pattern,''), entry_point, config, fire_count, last_fire_at, COALESCE(last_error,''), created_at, updated_at FROM triggers WHERE package_id = ? AND slug = ? AND type = 'webhook'`, packageID, slug). Scan(&t.ID, &t.PackageID, &t.Type, &t.Enabled, &t.Slug, &t.Secret, &t.EventPattern, &t.EntryPoint, &cfgStr, &t.FireCount, database.SNT(&t.LastFireAt), &t.LastError, database.ST(&t.CreatedAt), database.ST(&t.UpdatedAt)) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, err } t.Config = json.RawMessage(cfgStr) return &t, nil } func (s *TriggerStore) DeleteForPackage(ctx context.Context, packageID string) error { _, err := DB.ExecContext(ctx, `DELETE FROM triggers WHERE package_id = ?`, packageID) return err } func (s *TriggerStore) LogExecution(ctx context.Context, log *models.TriggerLog) error { _, err := DB.ExecContext(ctx, `INSERT INTO trigger_logs (id, trigger_id, scheduled_task_id, fired_at, duration_ms, success, error, output) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, uuid.New().String(), nullIfEmpty(log.TriggerID), nullIfEmpty(log.ScheduledTaskID), log.FiredAt, log.DurationMs, boolToInt(log.Success), nullIfEmpty(log.Error), nullIfEmpty(log.Output)) return err } func (s *TriggerStore) ListLogs(ctx context.Context, triggerID string, limit int) ([]models.TriggerLog, error) { if limit <= 0 { limit = 50 } rows, err := DB.QueryContext(ctx, `SELECT id, COALESCE(trigger_id,''), COALESCE(scheduled_task_id,''), fired_at, duration_ms, success, COALESCE(error,''), COALESCE(output,'') FROM trigger_logs WHERE trigger_id = ? ORDER BY fired_at DESC LIMIT ?`, triggerID, limit) if err != nil { return nil, err } defer rows.Close() return scanLogRowsSqlite(rows) } func (s *TriggerStore) PruneLogs(ctx context.Context, before time.Time) (int64, error) { result, err := DB.ExecContext(ctx, `DELETE FROM trigger_logs WHERE fired_at < ?`, before.UTC().Format(time.RFC3339)) if err != nil { return 0, err } return result.RowsAffected() } // ── Helpers ────────────────────────────────── func scanTriggerRowsSqlite(rows *sql.Rows) ([]models.Trigger, error) { var triggers []models.Trigger for rows.Next() { var t models.Trigger var cfgStr string if err := rows.Scan(&t.ID, &t.PackageID, &t.Type, &t.Enabled, &t.Slug, &t.EventPattern, &t.EntryPoint, &cfgStr, &t.FireCount, database.SNT(&t.LastFireAt), &t.LastError, database.ST(&t.CreatedAt), database.ST(&t.UpdatedAt)); err != nil { return nil, err } t.Config = json.RawMessage(cfgStr) triggers = append(triggers, t) } if triggers == nil { triggers = []models.Trigger{} } return triggers, nil } func scanLogRowsSqlite(rows *sql.Rows) ([]models.TriggerLog, error) { var logs []models.TriggerLog for rows.Next() { var l models.TriggerLog if err := rows.Scan(&l.ID, &l.TriggerID, &l.ScheduledTaskID, &l.FiredAt, &l.DurationMs, &l.Success, &l.Error, &l.Output); err != nil { return nil, err } logs = append(logs, l) } if logs == nil { logs = []models.TriggerLog{} } return logs, nil } // boolToInt and nullIfEmpty are defined in workflows.go (same package).