package db import ( "context" "database/sql" "fmt" "strings" ) // Insight represents a row in the insights table. type Insight struct { ID int64 `json:"id"` Type string `json:"type"` DateFrom string `json:"date_from"` DateTo string `json:"date_to"` Project *string `json:"project"` Agent string `json:"agent"` Model *string `json:"model"` Prompt *string `json:"prompt"` Content string `json:"content"` Kind string `json:"kind,omitempty"` SchemaVersion string `json:"schema_version,omitempty"` TemplateID string `json:"template_id,omitempty"` TemplateVersion string `json:"template_version,omitempty"` AggregateHash string `json:"aggregate_hash,omitempty"` CacheKey string `json:"cache_key,omitempty"` CacheStatus string `json:"cache_status,omitempty"` ProvenanceJSON string `json:"provenance_json,omitempty"` StructuredJSON string `json:"structured_json,omitempty"` CreatedAt string `json:"created_at"` } // InsightFilter specifies how to query insights. type InsightFilter struct { Type string // "daily_activity" or "agent_analysis" Project string // "" = no filter GlobalOnly bool // true = project IS NULL only DateFrom string // YYYY-MM-DD, "" = no filter DateTo string // YYYY-MM-DD, "" = no filter } const insightBaseCols = `id, type, date_from, date_to, project, agent, model, prompt, content, kind, schema_version, template_id, template_version, aggregate_hash, cache_key, cache_status, provenance_json, structured_json, created_at` func scanInsightRow(rs rowScanner) (Insight, error) { var s Insight err := rs.Scan( &s.ID, &s.Type, &s.DateFrom, &s.DateTo, &s.Project, &s.Agent, &s.Model, &s.Prompt, &s.Content, &s.Kind, &s.SchemaVersion, &s.TemplateID, &s.TemplateVersion, &s.AggregateHash, &s.CacheKey, &s.CacheStatus, &s.ProvenanceJSON, &s.StructuredJSON, &s.CreatedAt, ) return s, err } func buildInsightFilter( f InsightFilter, ) (string, []any) { var preds []string var args []any if f.Type != "" { preds = append(preds, "type = ?") args = append(args, f.Type) } if f.GlobalOnly { preds = append(preds, "project IS NULL") } else if f.Project != "" { preds = append(preds, "project = ?") args = append(args, f.Project) } if f.DateFrom != "" { preds = append(preds, "date_from >= ?") args = append(args, f.DateFrom) } if f.DateTo != "" { preds = append(preds, "date_to <= ?") args = append(args, f.DateTo) } if len(preds) == 0 { return "1=1", nil } return strings.Join(preds, " AND "), args } // InsertInsight inserts an insight and returns its ID. func (db *DB) InsertInsight(s Insight) (int64, error) { db.mu.Lock() defer db.mu.Unlock() res, err := db.getWriter().Exec(` INSERT INTO insights ( type, date_from, date_to, project, agent, model, prompt, content, kind, schema_version, template_id, template_version, aggregate_hash, cache_key, cache_status, provenance_json, structured_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, s.Type, s.DateFrom, s.DateTo, s.Project, s.Agent, s.Model, s.Prompt, s.Content, s.Kind, s.SchemaVersion, s.TemplateID, s.TemplateVersion, s.AggregateHash, s.CacheKey, s.CacheStatus, s.ProvenanceJSON, s.StructuredJSON, ) if err != nil { return 0, fmt.Errorf("inserting insight: %w", err) } return res.LastInsertId() } // GetCachedInsight returns the newest insight saved with cacheKey. // Returns nil, nil if no cache entry exists. func (db *DB) GetCachedInsight( ctx context.Context, cacheKey string, ) (*Insight, error) { if strings.TrimSpace(cacheKey) == "" { return nil, nil } row := db.getReader().QueryRowContext( ctx, "SELECT "+insightBaseCols+ " FROM insights WHERE cache_key = ?"+ " ORDER BY created_at DESC, id DESC LIMIT 1", cacheKey, ) s, err := scanInsightRow(row) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, fmt.Errorf( "getting cached insight: %w", err, ) } return &s, nil } const maxInsights = 500 // ListInsights returns insights matching the filter, // ordered by created_at DESC, capped at 500 rows. func (db *DB) ListInsights( ctx context.Context, f InsightFilter, ) ([]Insight, error) { where, args := buildInsightFilter(f) query := "SELECT " + insightBaseCols + " FROM insights WHERE " + where + " ORDER BY created_at DESC, id DESC" + " LIMIT " + fmt.Sprintf("%d", maxInsights) rows, err := db.getReader().QueryContext(ctx, query, args...) if err != nil { return nil, fmt.Errorf("querying insights: %w", err) } defer rows.Close() var insights []Insight for rows.Next() { s, err := scanInsightRow(rows) if err != nil { return nil, fmt.Errorf("scanning insight: %w", err) } insights = append(insights, s) } return insights, rows.Err() } // GetInsight returns a single insight by ID. // Returns nil, nil if not found. func (db *DB) GetInsight( ctx context.Context, id int64, ) (*Insight, error) { row := db.getReader().QueryRowContext( ctx, "SELECT "+insightBaseCols+ " FROM insights WHERE id = ?", id, ) s, err := scanInsightRow(row) if err == sql.ErrNoRows { return nil, nil } if err != nil { return nil, fmt.Errorf( "getting insight %d: %w", id, err, ) } return &s, nil } // CopyInsightsFrom copies all insights from the database at // sourcePath into this database using ATTACH/DETACH. func (db *DB) CopyInsightsFrom(sourcePath string) error { db.mu.Lock() defer db.mu.Unlock() // Pin a single connection for the ATTACH/INSERT/DETACH // sequence. database/sql's pool doesn't guarantee the // same underlying connection across separate Exec calls, // and ATTACH is connection-scoped. ctx := context.Background() conn, err := db.getWriter().Conn(ctx) if err != nil { return fmt.Errorf("acquiring connection: %w", err) } defer conn.Close() if _, err := conn.ExecContext( ctx, "ATTACH DATABASE ? AS old_db", sourcePath, ); err != nil { return fmt.Errorf("attaching source db: %w", err) } defer func() { _, _ = conn.ExecContext( ctx, "DETACH DATABASE old_db", ) }() hasCol := func(name string) bool { var count int err := conn.QueryRowContext(ctx, `SELECT count(*) FROM old_db.pragma_table_info('insights') WHERE name = ?`, name, ).Scan(&count) return err == nil && count > 0 } colExpr := func(name string) string { if hasCol(name) { return "COALESCE(" + name + ", '')" } return "''" } _, err = conn.ExecContext(ctx, ` INSERT OR IGNORE INTO insights (type, date_from, date_to, project, agent, model, prompt, content, kind, schema_version, template_id, template_version, aggregate_hash, cache_key, cache_status, provenance_json, structured_json, created_at) SELECT type, date_from, date_to, project, agent, model, prompt, content, `+colExpr("kind")+`, `+colExpr("schema_version")+`, `+colExpr("template_id")+`, `+colExpr("template_version")+`, `+colExpr("aggregate_hash")+`, `+colExpr("cache_key")+`, `+colExpr("cache_status")+`, `+colExpr("provenance_json")+`, `+colExpr("structured_json")+`, created_at FROM old_db.insights`) if err != nil { return fmt.Errorf("copying insights: %w", err) } return nil } // DeleteInsight removes an insight by ID. func (db *DB) DeleteInsight(id int64) error { db.mu.Lock() defer db.mu.Unlock() _, err := db.getWriter().Exec( "DELETE FROM insights WHERE id = ?", id, ) return err }