core/server/store/postgres/memory.go

package postgres

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

	"git.gobha.me/xcaliber/chat-switchboard/models"
	"git.gobha.me/xcaliber/chat-switchboard/store"
)

type MemoryStore struct{}

func NewMemoryStore() *MemoryStore { return &MemoryStore{} }

func (s *MemoryStore) Upsert(ctx context.Context, m *models.Memory) error {
	if m.ID == "" {
		m.ID = store.NewID()
	}
	if m.Status == "" {
		m.Status = models.MemoryStatusActive
	}
	if m.Confidence == 0 {
		m.Confidence = 1.0
	}

	_, err := DB.ExecContext(ctx, `
		INSERT INTO memories (id, scope, owner_id, user_id, key, value,
		                      source_channel_id, confidence, status)
		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
		ON CONFLICT (scope, owner_id, COALESCE(user_id, '00000000-0000-0000-0000-000000000000'), key)
		DO UPDATE SET
			value             = EXCLUDED.value,
			confidence        = EXCLUDED.confidence,
			status            = EXCLUDED.status,
			source_channel_id = EXCLUDED.source_channel_id,
			updated_at        = now()
	`, m.ID, m.Scope, m.OwnerID, m.UserID, m.Key, m.Value,
		m.SourceChannelID, m.Confidence, m.Status)
	return err
}

func (s *MemoryStore) Update(ctx context.Context, m *models.Memory) error {
	_, err := DB.ExecContext(ctx, `
		UPDATE memories
		SET key = $1, value = $2, confidence = $3, status = $4,
		    source_channel_id = $5, updated_at = now()
		WHERE id = $6
	`, m.Key, m.Value, m.Confidence, m.Status, m.SourceChannelID, m.ID)
	return err
}

func (s *MemoryStore) GetByID(ctx context.Context, id string) (*models.Memory, error) {
	m := &models.Memory{}
	err := DB.QueryRowContext(ctx, `
		SELECT id, scope, owner_id, user_id, key, value,
		       source_channel_id, confidence, status, created_at, updated_at
		FROM memories WHERE id = $1
	`, id).Scan(
		&m.ID, &m.Scope, &m.OwnerID, &m.UserID, &m.Key, &m.Value,
		&m.SourceChannelID, &m.Confidence, &m.Status, &m.CreatedAt, &m.UpdatedAt,
	)
	if err == sql.ErrNoRows {
		return nil, fmt.Errorf("memory not found: %s", id)
	}
	return m, err
}

func (s *MemoryStore) List(ctx context.Context, f models.MemoryFilter) ([]models.Memory, error) {
	clauses := []string{"scope = $1", "owner_id = $2"}
	args := []interface{}{f.Scope, f.OwnerID}
	idx := 3

	if f.UserID != nil {
		clauses = append(clauses, fmt.Sprintf("user_id = $%d", idx))
		args = append(args, *f.UserID)
		idx++
	}

	status := f.Status
	if status == "" {
		status = models.MemoryStatusActive
	}
	clauses = append(clauses, fmt.Sprintf("status = $%d", idx))
	args = append(args, status)
	idx++

	if f.Query != "" {
		clauses = append(clauses, fmt.Sprintf(
			"to_tsvector('english', key || ' ' || value) @@ plainto_tsquery('english', $%d)", idx))
		args = append(args, f.Query)
		idx++
	}

	limit := f.Limit
	if limit <= 0 || limit > 200 {
		limit = 50
	}
	args = append(args, limit)

	query := fmt.Sprintf(`
		SELECT id, scope, owner_id, user_id, key, value,
		       source_channel_id, confidence, status, created_at, updated_at
		FROM memories
		WHERE %s
		ORDER BY confidence DESC, updated_at DESC
		LIMIT $%d
	`, strings.Join(clauses, " AND "), idx)

	rows, err := DB.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	return scanMemories(rows)
}

func (s *MemoryStore) Delete(ctx context.Context, id string) error {
	_, err := DB.ExecContext(ctx, `DELETE FROM memories WHERE id = $1`, id)
	return err
}

func (s *MemoryStore) Archive(ctx context.Context, id string) error {
	_, err := DB.ExecContext(ctx, `
		UPDATE memories SET status = 'archived', updated_at = now() WHERE id = $1
	`, id)
	return err
}

func (s *MemoryStore) Recall(ctx context.Context, userID string, personaID *string, query string, limit int) ([]models.Memory, error) {
	if limit <= 0 || limit > 100 {
		limit = 20
	}

	// Build a UNION query across applicable scopes.
	// Always include user-scope memories for this user.
	// If a persona is active, also include persona and persona+user memories.

	parts := []string{}
	args := []interface{}{}
	idx := 1

	// 1. User-scope memories
	userPart := fmt.Sprintf(`
		SELECT id, scope, owner_id, user_id, key, value,
		       source_channel_id, confidence, status, created_at, updated_at
		FROM memories
		WHERE scope = 'user' AND owner_id = $%d AND status = 'active'
	`, idx)
	args = append(args, userID)
	idx++

	if query != "" {
		userPart += fmt.Sprintf(
			" AND to_tsvector('english', key || ' ' || value) @@ plainto_tsquery('english', $%d)", idx)
		args = append(args, query)
		idx++
	}
	parts = append(parts, userPart)

	// 2. Persona-scope memories (if persona active)
	if personaID != nil && *personaID != "" {
		personaPart := fmt.Sprintf(`
			SELECT id, scope, owner_id, user_id, key, value,
			       source_channel_id, confidence, status, created_at, updated_at
			FROM memories
			WHERE scope = 'persona' AND owner_id = $%d AND status = 'active'
		`, idx)
		args = append(args, *personaID)
		idx++
		if query != "" {
			personaPart += fmt.Sprintf(
				" AND to_tsvector('english', key || ' ' || value) @@ plainto_tsquery('english', $%d)", idx)
			args = append(args, query)
			idx++
		}
		parts = append(parts, personaPart)

		// 3. Persona+user memories
		puPart := fmt.Sprintf(`
			SELECT id, scope, owner_id, user_id, key, value,
			       source_channel_id, confidence, status, created_at, updated_at
			FROM memories
			WHERE scope = 'persona_user' AND owner_id = $%d AND user_id = $%d AND status = 'active'
		`, idx, idx+1)
		args = append(args, *personaID, userID)
		idx += 2
		if query != "" {
			puPart += fmt.Sprintf(
				" AND to_tsvector('english', key || ' ' || value) @@ plainto_tsquery('english', $%d)", idx)
			args = append(args, query)
			idx++
		}
		parts = append(parts, puPart)
	}

	// Scope priority: persona_user > persona > user (CASE ordering)
	fullQuery := fmt.Sprintf(`
		SELECT * FROM (%s) AS combined
		ORDER BY
			CASE scope
				WHEN 'persona_user' THEN 0
				WHEN 'persona'      THEN 1
				WHEN 'user'         THEN 2
			END,
			confidence DESC,
			updated_at DESC
		LIMIT $%d
	`, strings.Join(parts, " UNION ALL "), idx)
	args = append(args, limit)

	rows, err := DB.QueryContext(ctx, fullQuery, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	return scanMemories(rows)
}

func (s *MemoryStore) CountByOwner(ctx context.Context, scope, ownerID string) (int, error) {
	var count int
	err := DB.QueryRowContext(ctx, `
		SELECT COUNT(*) FROM memories
		WHERE scope = $1 AND owner_id = $2 AND status = 'active'
	`, scope, ownerID).Scan(&count)
	return count, err
}

func (s *MemoryStore) ListByStatus(ctx context.Context, status string, limit int) ([]models.Memory, error) {
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	rows, err := DB.QueryContext(ctx, `
		SELECT id, scope, owner_id, user_id, key, value,
		       source_channel_id, confidence, status, created_at, updated_at
		FROM memories
		WHERE status = $1
		ORDER BY created_at DESC
		LIMIT $2
	`, status, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	return scanMemories(rows)
}

// ── Helpers ─────────────────────────────────

func scanMemories(rows *sql.Rows) ([]models.Memory, error) {
	var memories []models.Memory
	for rows.Next() {
		var m models.Memory
		if err := rows.Scan(
			&m.ID, &m.Scope, &m.OwnerID, &m.UserID, &m.Key, &m.Value,
			&m.SourceChannelID, &m.Confidence, &m.Status, &m.CreatedAt, &m.UpdatedAt,
		); err != nil {
			return nil, err
		}
		memories = append(memories, m)
	}
	if memories == nil {
		memories = []models.Memory{}
	}
	return memories, rows.Err()
}

// ensure compile-time interface satisfaction
var _ store.MemoryStore = (*MemoryStore)(nil)

// ── CS5b additions (v0.29.0) ────────────────────────────────────────────

func (s *MemoryStore) SetEmbedding(ctx context.Context, id, embedding string) error {
	_, err := DB.ExecContext(ctx,
		`UPDATE memories SET embedding = $1::vector WHERE id = $2`, embedding, id)
	return err
}

func (s *MemoryStore) GetLastExtractionMessageID(ctx context.Context, channelID, userID string) (string, error) {
	var lastID string
	err := DB.QueryRowContext(ctx,
		`SELECT last_message_id FROM memory_extraction_log WHERE channel_id = $1 AND user_id = $2`,
		channelID, userID).Scan(&lastID)
	if err != nil {
		return "", nil // no entry yet
	}
	return lastID, nil
}

func (s *MemoryStore) UpsertExtractionLog(ctx context.Context, channelID, userID, lastMessageID string, count int) error {
	_, err := DB.ExecContext(ctx, `
		INSERT INTO memory_extraction_log (channel_id, user_id, last_message_id, memory_count)
		VALUES ($1, $2, $3, $4)
		ON CONFLICT(channel_id, user_id) DO UPDATE SET
			last_message_id = EXCLUDED.last_message_id,
			extracted_at = now(),
			memory_count = memory_extraction_log.memory_count + EXCLUDED.memory_count
	`, channelID, userID, lastMessageID, count)
	return err
}

// ── v0.30.1 — Memory Compaction ────────────────────────────────────────

func (s *MemoryStore) DecayConfidence(ctx context.Context, olderThan string, decayFactor float64) (int, error) {
	res, err := DB.ExecContext(ctx, `
		UPDATE memories
		SET confidence = confidence * $1,
		    updated_at = now()
		WHERE status = 'active'
		  AND updated_at < $2::timestamptz
		  AND confidence > 0.1
	`, decayFactor, olderThan)
	if err != nil {
		return 0, err
	}
	n, _ := res.RowsAffected()
	return int(n), nil
}

func (s *MemoryStore) Prune(ctx context.Context, belowConfidence float64) (int, error) {
	res, err := DB.ExecContext(ctx, `
		UPDATE memories
		SET status = 'archived',
		    updated_at = now()
		WHERE status = 'active'
		  AND confidence < $1
	`, belowConfidence)
	if err != nil {
		return 0, err
	}
	n, _ := res.RowsAffected()
	return int(n), nil
}