kenn-io--agentsview/internal/postgres/curation.go

package postgres

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"go.kenn.io/agentsview/internal/db"
)

// StarSession marks a session as starred in the shared PG dashboard
// metadata. Returns false when the session does not exist.
func (s *Store) StarSession(sessionID string) (bool, error) {
	res, err := s.pg.Exec(`
		INSERT INTO starred_sessions (session_id)
		SELECT $1 WHERE EXISTS (
			SELECT 1 FROM sessions WHERE id = $1
		)
		ON CONFLICT (session_id) DO NOTHING`,
		sessionID,
	)
	if err != nil {
		return false, fmt.Errorf("starring session %s: %w", sessionID, err)
	}
	n, _ := res.RowsAffected()
	if n > 0 {
		return true, nil
	}

	var exists int
	err = s.pg.QueryRow(
		`SELECT 1 FROM sessions WHERE id = $1`,
		sessionID,
	).Scan(&exists)
	if err == sql.ErrNoRows {
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("checking session %s: %w", sessionID, err)
	}
	return true, nil
}

// UnstarSession removes a session star from the shared PG dashboard
// metadata.
func (s *Store) UnstarSession(sessionID string) error {
	if _, err := s.pg.Exec(
		`DELETE FROM starred_sessions WHERE session_id = $1`,
		sessionID,
	); err != nil {
		return fmt.Errorf("unstarring session %s: %w", sessionID, err)
	}
	return nil
}

// ListStarredSessionIDs returns shared PG-starred session IDs.
func (s *Store) ListStarredSessionIDs(
	ctx context.Context,
) ([]string, error) {
	rows, err := s.pg.QueryContext(ctx, `
		SELECT session_id
		FROM starred_sessions
		ORDER BY created_at DESC, session_id DESC`)
	if err != nil {
		return nil, fmt.Errorf("listing starred sessions: %w", err)
	}
	defer rows.Close()

	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			return nil, fmt.Errorf("scanning starred session: %w", err)
		}
		ids = append(ids, id)
	}
	return ids, rows.Err()
}

// BulkStarSessions stars multiple existing sessions in one transaction.
// Unknown session IDs are skipped.
func (s *Store) BulkStarSessions(sessionIDs []string) error {
	if len(sessionIDs) == 0 {
		return nil
	}

	tx, err := s.pg.Begin()
	if err != nil {
		return fmt.Errorf("beginning star transaction: %w", err)
	}
	defer func() { _ = tx.Rollback() }()

	stmt, err := tx.Prepare(`
		INSERT INTO starred_sessions (session_id)
		SELECT $1 WHERE EXISTS (
			SELECT 1 FROM sessions WHERE id = $1
		)
		ON CONFLICT (session_id) DO NOTHING`)
	if err != nil {
		return fmt.Errorf("preparing star statement: %w", err)
	}
	defer stmt.Close()

	for _, id := range sessionIDs {
		if _, err := stmt.Exec(id); err != nil {
			return fmt.Errorf("starring session %s: %w", id, err)
		}
	}

	return tx.Commit()
}

// PinMessage creates or updates a shared PG pin for a message. PG
// exposes message IDs as the message ordinal, matching scanPGMessages.
// On conflict, ordinal and source_uuid are refreshed from the current
// message so the pin tracks whatever is at message_id today;
// created_at is preserved by being absent from the SET clause.
func (s *Store) PinMessage(
	sessionID string, messageID int64, note *string,
) (int64, error) {
	var id int64
	err := s.pg.QueryRow(`
		WITH upsert AS (
			INSERT INTO pinned_messages (
				session_id, message_id, ordinal, source_uuid, note
			)
			SELECT $1, m.ordinal, m.ordinal,
				COALESCE(m.source_uuid, ''), $3
			FROM messages m
			WHERE m.session_id = $1 AND m.ordinal = $2
			ON CONFLICT (session_id, message_id)
			DO UPDATE SET
				note = EXCLUDED.note,
				ordinal = EXCLUDED.ordinal,
				source_uuid = EXCLUDED.source_uuid
			RETURNING id
		)
		SELECT id FROM upsert`,
		sessionID, messageID, note,
	).Scan(&id)
	if err == sql.ErrNoRows {
		return 0, nil
	}
	if err != nil {
		return 0, fmt.Errorf("pinning message: %w", err)
	}
	return id, nil
}

// UnpinMessage removes a shared PG pin.
func (s *Store) UnpinMessage(sessionID string, messageID int64) error {
	if _, err := s.pg.Exec(
		`DELETE FROM pinned_messages
		 WHERE session_id = $1 AND message_id = $2`,
		sessionID, messageID,
	); err != nil {
		return fmt.Errorf("unpinning message: %w", err)
	}
	return nil
}

// ListPinnedMessages returns shared PG pins, optionally filtered by
// session or project. All-pins results include message and session
// metadata for the pinned page.
func (s *Store) ListPinnedMessages(
	ctx context.Context, sessionID string, project string,
) ([]db.PinnedMessage, error) {
	var (
		rows *sql.Rows
		err  error
	)
	if sessionID != "" {
		rows, err = s.pg.QueryContext(ctx, `
			SELECT id, session_id, message_id, ordinal, note, created_at
			FROM pinned_messages
			WHERE session_id = $1
			ORDER BY created_at DESC, id DESC`,
			sessionID,
		)
	} else {
		query := `
			SELECT p.id, p.session_id, p.message_id, p.ordinal,
				p.note, p.created_at, m.content, m.role,
				s.project, s.agent, COALESCE(s.display_name, s.session_name), s.first_message
			FROM pinned_messages p
			JOIN sessions s
				ON p.session_id = s.id
				AND s.deleted_at IS NULL
			LEFT JOIN messages m
				ON m.session_id = p.session_id
				AND m.ordinal = p.message_id`
		args := []any{}
		if project != "" {
			query += " WHERE s.project = $1"
			args = append(args, project)
		}
		query += " ORDER BY p.created_at DESC, p.id DESC LIMIT 500"
		rows, err = s.pg.QueryContext(ctx, query, args...)
	}
	if err != nil {
		return nil, fmt.Errorf("listing pinned messages: %w", err)
	}
	defer rows.Close()

	var pins []db.PinnedMessage
	withContent := sessionID == ""
	for rows.Next() {
		p, scanErr := scanPGPinnedMessage(rows, withContent)
		if scanErr != nil {
			return nil, scanErr
		}
		pins = append(pins, p)
	}
	return pins, rows.Err()
}

func scanPGPinnedMessage(
	rs interface{ Scan(dest ...any) error },
	withContent bool,
) (db.PinnedMessage, error) {
	var p db.PinnedMessage
	var createdAt time.Time
	var err error
	if withContent {
		err = rs.Scan(
			&p.ID, &p.SessionID, &p.MessageID,
			&p.Ordinal, &p.Note, &createdAt,
			&p.Content, &p.Role,
			&p.SessionProject, &p.SessionAgent,
			&p.SessionDisplayName, &p.SessionFirstMessage,
		)
	} else {
		err = rs.Scan(
			&p.ID, &p.SessionID, &p.MessageID,
			&p.Ordinal, &p.Note, &createdAt,
		)
	}
	if err != nil {
		return p, fmt.Errorf("scanning pinned message: %w", err)
	}
	p.CreatedAt = FormatISO8601(createdAt)
	return p, nil
}