// No `unlink` here, deliberately: the backend mounts /music `:ro` and must never
// remove a library file. Deletions are queued into pending_file_deletions and
// carried out by the worker's gated cleanup sweep.
import { readFile } from 'fs/promises';
import { fileURLToPath } from 'url';
import { dirname, join } from 'path';
import { Pool, PoolClient } from 'pg';
import { SearchService } from './search.service.js';

/** Anything with a `.query()` — either the shared Pool or a checked-out client. */
type Queryable = Pool | PoolClient;

function calendarContextKey(context: Record<string, unknown> | null | undefined): string | null {
  const hour = context?.localHour;
  const weekday = context?.weekday;
  const month = context?.month;
  if (typeof hour !== 'number' || typeof weekday !== 'number' || typeof month !== 'number'
    || !Number.isInteger(hour) || !Number.isInteger(weekday) || !Number.isInteger(month)
    || hour < 0 || hour > 23 || weekday < 0 || weekday > 6 || month < 1 || month > 12) return null;
  const daypart = hour < 6 ? 'night' : hour < 12 ? 'morning' : hour < 18 ? 'day' : 'evening';
  const dayType = weekday === 0 || weekday === 6 ? 'weekend' : 'weekday';
  const season = month === 12 || month <= 2 ? 'winter' : month <= 5 ? 'spring' : month <= 8 ? 'summer' : 'autumn';
  return `calendar:${daypart}:${dayType}:${season}`;
}

type VibePlanVersionRow = Omit<VibePlan, 'items'> & {
  item_plan_version_id: string | null;
  ordinal: number | null;
  track_id: string | null;
  slot_role: string | null;
  candidate_source: string | null;
  score: number | null;
  score_breakdown: Record<string, unknown> | null;
  explanation: unknown | null;
  committed: boolean | null;
};

import { MIGRATIONS } from '../db/migrations.js';
import { allowedFields } from '../db/updatable-columns.js';

// Row shapes live in ../db/types.ts; re-exported here so existing importers
// (routes, generators, session-director) need no change.
import type {
  Artist,
  Album,
  TrackArtist,
  Track,
  FeedbackAction,
  HistoryEntry,
  Genre,
  DislikeEntry,
  ArtistWithAlbums,
  AlbumWithTracks,
  Claim,
  Evidence,
  ListenerBelief,
  ClaimEdge,
  SessionState,
  DiversityBudget,
  RepetitionRule,
  VibeSession,
  VibeSessionStatus,
  VibeEvent,
  RecordedVibeEvent,
  VibePlan,
  VibePlanItem,
  VibeSessionProfile,
} from '../db/types.js';
import { AUDIO_PREFERENCE_BUCKETS, FEEDBACK_ACTIONS } from '../db/types.js';
export * from '../db/types.js';

/**
 * Every play_history write goes through this statement.
 *
 * The title/artist copy is taken in the same round trip via LEFT JOIN, so a
 * play stays readable after its track is hard-deleted, and the join never
 * suppresses the insert when the track id is unknown.
 */
const PLAY_HISTORY_INSERT = `
  INSERT INTO play_history (
    user_id, track_id, batch_id, completed, listened_ms, track_title, track_artist
  )
  SELECT $1::uuid, $2::uuid, $3::uuid, $4::boolean, $5::int, t.title, t.artist
    FROM (SELECT 1) AS always
    LEFT JOIN tracks t ON t.id = $2::uuid
  RETURNING id`;

export class DbService {
  /** Exposed so route handlers (e.g. settings) can query the database directly. */
  readonly pgClient: Pool;

  constructor(
    pgClient: Pool,
    private searchService?: SearchService
  ) {
    this.pgClient = pgClient;
  }

  /**
   * Check out a dedicated client for a transaction and BEGIN/COMMIT/ROLLBACK
   * on it, then release it back to the pool. Never run BEGIN...COMMIT on
   * `this.pgClient` directly — the Pool hands out a different connection to
   * every query, so a shared-connection transaction would let concurrent
   * requests' queries land inside it (a ROLLBACK could discard another
   * request's writes). Pass `client` through to any nested calls that must
   * participate in the same transaction (e.g. recordEvidence, upsertClaim).
   */
  private async withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
    const client = await this.pgClient.connect();
    try {
      await client.query('BEGIN');
      const result = await fn(client);
      await client.query('COMMIT');
      return result;
    } catch (err) {
      await client.query('ROLLBACK');
      throw err;
    } finally {
      client.release();
    }
  }

  /**
   * Apply the canonical schema (backend/src/db/schema.sql) on boot. The file is
   * fully idempotent — enums are guarded with DO/EXCEPTION blocks and every
   * table/index uses IF NOT EXISTS — so running it on every startup is safe and
   * self-provisions tables on databases whose volume predates a schema change
   * (the docker-entrypoint-initdb.d mount only runs on FIRST init). This is what
   * keeps "relation \"play_history\"/\"feedback\" does not exist" from recurring.
   */
  async ensureSchema(): Promise<void> {
    // Resolve relative to this module. At runtime this is dist/services/, and
    // the SQL ships unbuilt at src/db/schema.sql (Dockerfile `COPY . .`), so go
    // up two levels from dist/services -> app root, then into src/db.
    const here = dirname(fileURLToPath(import.meta.url));
    const schemaPath = join(here, '..', '..', 'src', 'db', 'schema.sql');
    const sql = await readFile(schemaPath, 'utf8');
    await this.pgClient.query(sql);
  }

  /**
   * Run pending schema migrations on boot.
   *
   * Each migration is a { id, sql } object. `id` must be a stable, unique string
   * (convention: "YYYYMMDD_short_description"). Once applied, the id is recorded
   * in `schema_migrations` and never re-run — even if the SQL changes.
   *
   * To add a new migration: append to the MIGRATIONS array below. Never edit or
   * remove an existing entry — that would leave the migration "applied" in the DB
   * but with different SQL in code, which is a lie. Instead, add a new migration.
   */
  async runMigrations(): Promise<void> {
    await this.pgClient.query(`
      CREATE TABLE IF NOT EXISTS schema_migrations (
        id TEXT PRIMARY KEY,
        applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
      )
    `);

    const applied = await this.pgClient.query('SELECT id FROM schema_migrations');
    const appliedIds = new Set(applied.rows.map((r: any) => r.id as string));

    for (const migration of MIGRATIONS) {
      if (appliedIds.has(migration.id)) continue;
      console.log(`[DB] Running migration: ${migration.id}`);
      try {
        await this.withTransaction(async (client) => {
          await client.query(migration.sql);
          await client.query('INSERT INTO schema_migrations (id) VALUES ($1)', [migration.id]);
        });
        console.log(`[DB] Migration applied: ${migration.id}`);
      } catch (err) {
        console.error(`[DB] Migration failed: ${migration.id}`, err);
        throw err;
      }
    }
  }

  async getTracks(params: { limit?: number; offset?: number; sort_by?: string; order?: 'ASC' | 'DESC'; search?: string } = {}) {
    const { limit = 100, offset = 0, sort_by = 'title', order = 'ASC', search } = params;
    const validSortBy = ['title', 'artist', 'album_id', 'duration', 'play_count'];
    const sortColumn = validSortBy.includes(sort_by) ? `"${sort_by}"` : '"title"';
    const sortOrder = order === 'DESC' ? 'DESC' : 'ASC';

    if (search && this.searchService) {
      const searchResults = await this.searchService.search('tracks', search, {
        query_by: 'title,artist'
      });
      return (searchResults?.hits || []).map((h: any) => h.document) as Track[];
    }

    // Exclude HIDDEN (disliked) and DELETED tracks from library views.
    // Join albums to get artwork_id for track cover display.
    let query = `
      SELECT t.*, al.artwork_id
      FROM tracks t
      LEFT JOIN albums al ON al.id = t.album_id
      WHERE t.state NOT IN ('HIDDEN', 'DELETED')
    `;
    const queryParams: any[] = [limit, offset];
    let paramIndex = 3;

    if (search) {
      query += ` AND (t.title ILIKE $${paramIndex} OR t.artist ILIKE $${paramIndex})`;
      queryParams.push(`%${search}%`);
    }

    query += ` ORDER BY ${sortColumn} ${sortOrder} LIMIT $1 OFFSET $2`;

    const res = await this.pgClient.query(query, queryParams);
    return this.attachArtists(res.rows as Track[]);
  }

  /** Attach artists array (from track_artists join) to a list of tracks. */
  private async attachArtists(tracks: Track[]): Promise<Track[]> {
    if (tracks.length === 0) return tracks;
    const ids = tracks.map((t) => t.id);
    const artistsRes = await this.pgClient.query(
      `SELECT ta.track_id, ta.role, a.id, a.name
       FROM track_artists ta
       JOIN artists a ON a.id = ta.artist_id
       WHERE ta.track_id = ANY($1)
       ORDER BY ta.role DESC`,
      [ids]
    );
    const byTrack = new Map<string, TrackArtist[]>();
    for (const row of artistsRes.rows) {
      if (!byTrack.has(row.track_id)) byTrack.set(row.track_id, []);
      byTrack.get(row.track_id)!.push({ id: row.id, name: row.name, role: row.role });
    }
    return tracks.map((t) => ({ ...t, artists: byTrack.get(t.id) ?? [] }));
  }

  // Search returning a Typesense-compatible shape ({ found, hits:[{document}] }).
  // Prefers Typesense when its 'tracks' collection is populated; falls back to a
  // Postgres ILIKE scan when Typesense isn't indexed/reachable yet (no indexing
  // pipeline exists today — see TODO: build a tracks reindex into Typesense).
  async searchTracks(q: string, limit = 50): Promise<{ found: number; hits: { document: Track }[] }> {
    if (this.searchService) {
      try {
        const res: any = await this.searchService.search('tracks', q, {
          query_by: 'title,artist',
          per_page: limit,
        });
        // Only trust Typesense when it actually returns hits. An empty result is
        // ambiguous: it usually means the 'tracks' collection is unindexed (no
        // indexing pipeline runs on scan), not that there are genuinely no
        // matches — so fall through to the Postgres scan instead of returning [].
        if (res && Array.isArray(res.hits) && res.hits.length > 0) {
          return {
            found: res.found ?? res.hits.length,
            hits: res.hits.map((h: any) => ({ document: h.document as Track })),
          };
        }
      } catch {
        // Typesense collection missing/unreachable — fall through to Postgres.
      }
    }

    const like = `%${q}%`;
    const res = await this.pgClient.query(
      `SELECT t.*, al.artwork_id FROM tracks t
       LEFT JOIN albums al ON al.id = t.album_id
       WHERE t.state = 'LIBRARY' AND (t.title ILIKE $1 OR t.artist ILIKE $1)
       ORDER BY t.play_count DESC, t.title ASC
       LIMIT $2`,
      [like, limit]
    );
    const rows = res.rows as Track[];
    return { found: rows.length, hits: rows.map((t) => ({ document: t })) };
  }

  async getArtists(params: { limit?: number; offset?: number } = {}): Promise<Artist[]> {
    const { limit = 50, offset = 0 } = params;
    const res = await this.pgClient.query(
      'SELECT * FROM artists ORDER BY name ASC LIMIT $1 OFFSET $2',
      [limit, offset]
    );
    return res.rows as Artist[];
  }

  async getArtistsById(id: string): Promise<ArtistWithAlbums | null> {
    const artistRes = await this.pgClient.query('SELECT * FROM artists WHERE id = $1', [id]);
    const artist = artistRes.rows[0] as Artist;
    if (!artist) return null;

    const albumsRes = await this.pgClient.query('SELECT * FROM albums WHERE artist_id = $1', [id]);
    const albums = albumsRes.rows as Album[];

    return {
      ...artist,
      albums,
    };
  }

  async getSimilarArtists(artistId: string): Promise<{ similar_name: string; match: number }[]> {
    const res = await this.pgClient.query(
      `SELECT similar_name, match
       FROM artist_similar
       WHERE artist_id = $1
       ORDER BY match DESC`,
      [artistId]
    );
    return res.rows;
  }

  async getAlbums(params: { limit?: number; offset?: number } = {}): Promise<Album[]> {
    const { limit = 50, offset = 0 } = params;
    const res = await this.pgClient.query(
      `SELECT al.*, ar.name AS artist_name
       FROM albums al
       LEFT JOIN artists ar ON ar.id = al.artist_id
       ORDER BY al.title ASC
       LIMIT $1 OFFSET $2`,
      [limit, offset]
    );
    return res.rows as Album[];
  }

  async getAlbumById(id: string): Promise<AlbumWithTracks | null> {
    const albumRes = await this.pgClient.query('SELECT * FROM albums WHERE id = $1', [id]);
    const album = albumRes.rows[0] as Album;
    if (!album) return null;

    const tracksRes = await this.pgClient.query(
      `SELECT t.*, al.artwork_id FROM tracks t
       LEFT JOIN albums al ON al.id = t.album_id
       WHERE t.album_id = $1 ORDER BY t.title ASC`, [id]);
    const tracks = await this.attachArtists(tracksRes.rows as Track[]);

    return {
      ...album,
      tracks,
    };
  }

  /**
   * Library totals for the Home header. Counts only what library views show,
   * so the number on screen matches what a user can actually browse.
   */
  async getLibraryStats(): Promise<{ tracks: number; albums: number; artists: number; duration: number }> {
    const res = await this.pgClient.query(
      `SELECT
         (SELECT COUNT(*)::int FROM tracks WHERE state NOT IN ('HIDDEN','DELETED')) AS tracks,
         (SELECT COUNT(*)::int FROM albums) AS albums,
         (SELECT COUNT(*)::int FROM artists) AS artists,
         (SELECT COALESCE(SUM(duration),0)::int FROM tracks WHERE state NOT IN ('HIDDEN','DELETED')) AS duration`
    );
    return res.rows[0];
  }

  async getGenres(): Promise<Genre[]> {
    const res = await this.pgClient.query(
      `SELECT g.id, g.name, g.parent_id, COUNT(tg.track_id)::int AS track_count
       FROM genre g
       LEFT JOIN track_genre tg ON tg.genre_id = g.id
       GROUP BY g.id
       ORDER BY track_count DESC, g.name`
    );
    return res.rows as Genre[];
  }

  async getGenreById(id: string): Promise<Genre | null> {
    const res = await this.pgClient.query(
      `SELECT g.id, g.name, g.parent_id, COUNT(tg.track_id)::int AS track_count
       FROM genre g
       LEFT JOIN track_genre tg ON tg.genre_id = g.id
       WHERE g.id = $1
       GROUP BY g.id`,
      [id]
    );
    return (res.rows[0] as Genre) || null;
  }

  async getTracksByGenre(genreId: string, limit = 100, offset = 0): Promise<Track[]> {
    const res = await this.pgClient.query(
      `SELECT t.id, t.path, t.hash, t.title, t.artist, t.album_id, t.duration,
              t.state, t.play_count, t.skip_count, t.dislike_count,
              t.last_played_at, t.mtime, t.source_type, al.artwork_id
         FROM tracks t
         JOIN track_genre tg ON tg.track_id = t.id
         LEFT JOIN albums al ON al.id = t.album_id
         WHERE tg.genre_id = $1 AND t.state = 'LIBRARY'
         ORDER BY tg.weight DESC
         LIMIT $2 OFFSET $3`,
      [genreId, limit, offset]
    );
    return res.rows as Track[];
  }

  async getFavorites(userId: string): Promise<Track[]> {
    // Favorites and disliked tracks are mutually exclusive — a disliked track
    // has state='HIDDEN' so it won't appear here. Defensive filter anyway.
    const query = `
      SELECT t.*, al.artwork_id FROM tracks t
      JOIN favorites f ON t.id = f.track_id
      LEFT JOIN albums al ON al.id = t.album_id
      WHERE f.user_id = $1 AND t.state = 'LIBRARY'
    `;
    const res = await this.pgClient.query(query, [userId]);
    return res.rows as Track[];
  }

  async addFavorite(userId: string, trackId: string): Promise<void> {
    await this.pgClient.query('INSERT INTO favorites (user_id, track_id) VALUES ($1, $2) ON CONFLICT DO NOTHING', [userId, trackId]);
  }

  async removeFavorite(userId: string, trackId: string): Promise<void> {
    await this.pgClient.query('DELETE FROM favorites WHERE user_id = $1 AND track_id = $2', [userId, trackId]);
  }

  /**
   * Dislike a track: atomically hides the track, inserts a dislike row, and
   * logs a feedback event. Per the lifecycle spec, the track transitions from
   * LIBRARY -> HIDDEN and is removed from all active views immediately.
   */
  async dislikeTrack(userId: string, trackId: string): Promise<void> {
    await this.withTransaction(async (client) => {
      // Phase 1: hide the track in all active views. A probation recommendation
      // is disliked the same way a library track is, and retires on the spot —
      // the listener has answered the question probation exists to ask.
      await client.query(
        `UPDATE tracks
            SET state = 'HIDDEN',
                probation_status = CASE WHEN state = 'RECOMMENDED' THEN 'retired' ELSE probation_status END
          WHERE id = $1 AND state IN ('LIBRARY', 'RECOMMENDED')`,
        [trackId]
      );

      // Phase 1: insert dislike row (idempotent — won't create duplicate)
      await client.query(
        'INSERT INTO dislikes (track_id) VALUES ($1) ON CONFLICT (track_id) DO NOTHING',
        [trackId]
      );

      // Phase 1: log feedback signal for the Vibe learning loop
      await client.query(
        "INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'disliked')",
        [userId, trackId]
      );
    });

    // Write evidence: hidden → negative profile (only on success), then carry
    // that signal through the track's artist/genre/audio identities.
    await this.recordTrackEvidence({
      user_id: userId,
      track_id: trackId,
      signal: 'hidden',
      profile: 'negative',
      weight: -0.60,
    });
  }

  /**
   * Restore a disliked track: atomically removes the dislike row and sets the
   * track back to LIBRARY. Per the lifecycle spec, this is the "User Recovery"
   * reversal of the dislike action.
   */
  async restoreDislike(trackId: string): Promise<void> {
    await this.withTransaction(async (client) => {
      await client.query('DELETE FROM dislikes WHERE track_id = $1', [trackId]);
      await client.query("UPDATE tracks SET state = 'LIBRARY' WHERE id = $1", [trackId]);
    });
  }

  /**
   * Fetch all currently disliked tracks (in HIDDEN state via dislike rows).
   * Returns DislikeEntry rows joined with track metadata. Used by the
   * Quarantine/dislikes list endpoint.
   */
  async getDislikedTracks(): Promise<DislikeEntry[]> {
    const res = await this.pgClient.query(
      `SELECT d.track_id, d.disliked_at, d.warned_at, d.deleted_at,
              d.grace_hours, d.state,
              t.title AS track_title, t.artist AS track_artist, t.path AS track_path
       FROM dislikes d
       JOIN tracks t ON t.id = d.track_id
       ORDER BY d.disliked_at DESC`
    );
    return res.rows as DislikeEntry[];
  }

  /**
   * Fetch a single dislike row by track_id, or null if not disliked.
   */
  async getDislikeByTrackId(trackId: string): Promise<DislikeEntry | null> {
    const res = await this.pgClient.query(
      `SELECT d.track_id, d.disliked_at, d.warned_at, d.deleted_at,
              d.grace_hours, d.state,
              t.title AS track_title, t.artist AS track_artist, t.path AS track_path
       FROM dislikes d
       JOIN tracks t ON t.id = d.track_id
       WHERE d.track_id = $1`,
      [trackId]
    );
    return (res.rows[0] as DislikeEntry) || null;
  }

  // NOTE: hardDeleteTrack() used to live here as a second, unreferenced deletion
  // path with the same insert-then-delete shape as permanentlyDeleteTrack(), but
  // with the audit insert outside any transaction and the unlink after the DELETE
  // and silently swallowed. It had no callers. Removed rather than fixed twice
  // over: there is now exactly one deletion path in the backend
  // (permanentlyDeleteTrack, below) and exactly one unlink site in the whole
  // system (workers/src/cleanup.service.ts).

  async recordPlay(
    userId: string,
    trackId: string,
    completed: boolean,
    batchId?: string,
    listenedMs?: number,
  ): Promise<string> {
    if (!completed) {
      const res = await this.pgClient.query(
        PLAY_HISTORY_INSERT,
        [userId, trackId, batchId ?? null, false, listenedMs ?? null]
      );
      return res.rows[0].id as string;
    }

    // Completed play: history insert + play_count bump + Success-Driven Center
    // + evidence writing + listener-behavior claims. All atomic.
    return this.withTransaction(async (client) => {
      // 1. Record play history
      const insertRes = await client.query(
        PLAY_HISTORY_INSERT,
        [userId, trackId, batchId ?? null, true, listenedMs ?? null]
      );
      const historyId = insertRes.rows[0].id as string;

      // 2. Bump play count + last_played_at
      await client.query(
        'UPDATE tracks SET play_count = play_count + 1, last_played_at = NOW() WHERE id = $1',
        [trackId]
      );

      // 3. Write evidence: playback_completed → longterm affinity
      await this.recordTrackEvidence({
        user_id: userId,
        track_id: trackId,
        signal: 'playback_completed',
        profile: 'longterm',
        weight: 0.10,
        context: batchId ? { batch_id: batchId } : undefined,
      }, client);

      // 4. Check for replay within 24h → strengthens longterm + obsession
      const recentPlays = await client.query(
        `SELECT COUNT(*)::int AS cnt FROM play_history
         WHERE user_id = $1 AND track_id = $2 AND completed = true
           AND played_at > NOW() - INTERVAL '24 hours'`,
        [userId, trackId]
      );
      if ((recentPlays.rows[0]?.cnt as number) > 1) {
        await this.recordTrackEvidence({
          user_id: userId,
          track_id: trackId,
          signal: 'replay_within_24h',
          profile: 'longterm',
          weight: 0.25,
        }, client);
        await this.recordTrackEvidence({
          user_id: userId,
          track_id: trackId,
          signal: 'replay_within_24h',
          profile: 'obsession',
          weight: 0.40,
        }, client);
      }

      // 5. Listener-behavior writer: back-to-back play within 30 min → weak edges
      //    Resolve artist IDs for current track and previous track, then write
      //    alias_of (same artist, different name) or same_scene_as (different artists).
      //    Reads track_artists_v2 (claim-fusion-backed), NOT the legacy
      //    track_artists table — the vibe engine reads claim_fusion, so writing
      //    claims keyed off a possibly-stale legacy table silently stops these
      //    edges from being written for newly-enriched tracks.
      const currentArtist = await client.query(
        `SELECT tv.artist_id, a.normalized_name
         FROM track_artists_v2 tv
         JOIN artists a ON a.id = tv.artist_id
         WHERE tv.track_id = $1 AND tv.role = 'main'
         ORDER BY tv.confidence DESC
         LIMIT 1`,
        [trackId]
      );
      const currentArtistRow = currentArtist.rows[0] as { artist_id: string; normalized_name: string } | undefined;

      if (currentArtistRow) {
        // Previous completed play's track + the gap in minutes, in one query
        // via LAG() instead of a separate prev-track lookup + gap lookup.
        const prevPlay = await client.query(
          `SELECT prev_track_id, EXTRACT(EPOCH FROM (played_at - prev_played_at)) / 60 AS min_gap
           FROM (
             SELECT track_id, played_at,
                    LAG(track_id) OVER (ORDER BY played_at) AS prev_track_id,
                    LAG(played_at) OVER (ORDER BY played_at) AS prev_played_at
             FROM play_history
             WHERE user_id = $1 AND completed = true
           ) w
           WHERE w.track_id = $2
           ORDER BY w.played_at DESC
           LIMIT 1`,
          [userId, trackId]
        );
        const prevTrackId = prevPlay.rows[0]?.prev_track_id as string | undefined;
        const gapMinutes = prevPlay.rows[0]?.min_gap as number | undefined;

        if (prevTrackId && gapMinutes !== undefined && gapMinutes <= 30) {
          const prevArtist = await client.query(
            `SELECT tv.artist_id, a.normalized_name
             FROM track_artists_v2 tv
             JOIN artists a ON a.id = tv.artist_id
             WHERE tv.track_id = $1 AND tv.role = 'main'
             ORDER BY tv.confidence DESC
             LIMIT 1`,
            [prevTrackId]
          );
          const prevArtistRow = prevArtist.rows[0] as { artist_id: string; normalized_name: string } | undefined;

          if (prevArtistRow) {
            if (prevArtistRow.normalized_name === currentArtistRow.normalized_name) {
              // Same normalized artist name → weak alias_of
              await this.upsertClaim({
                user_id: userId,
                subject_type: 'artist',
                subject_id: prevArtistRow.artist_id,
                predicate: 'alias_of',
                object_type: 'artist',
                object_id: currentArtistRow.artist_id,
                source: 'listener_behavior',
                confidence: 0.2,
              }, client);
            } else {
              // Different artists played back-to-back → weak same_scene_as
              await this.upsertClaim({
                user_id: userId,
                subject_type: 'artist',
                subject_id: prevArtistRow.artist_id,
                predicate: 'same_scene_as',
                object_type: 'artist',
                object_id: currentArtistRow.artist_id,
                source: 'listener_behavior',
                confidence: 0.3,
              }, client);
            }
          }
        }
      }

      return historyId;
    });
  }

  async recordSkip(userId: string, trackId: string): Promise<void> {
    // Skips do NOT move the center (transient per spec). Also writes negative evidence.
    await this.withTransaction(async (client) => {
      await client.query('UPDATE tracks SET skip_count = skip_count + 1 WHERE id = $1', [trackId]);
      await client.query(
        "INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, 'skipped')",
        [userId, trackId]
      );
      // Write evidence: skip_quick → negative profile
      await this.recordTrackEvidence({
        user_id: userId,
        track_id: trackId,
        signal: 'skip_quick',
        profile: 'negative',
        weight: -0.20,
      }, client);
    });
  }

  async recordFeedback(userId: string, trackId: string, action: FeedbackAction): Promise<void> {
    if (!FEEDBACK_ACTIONS.includes(action)) {
      throw new Error(`Invalid feedback action: ${action}`);
    }
    await this.pgClient.query(
      'INSERT INTO feedback (user_id, track_id, action) VALUES ($1, $2, $3)',
      [userId, trackId, action]
    );

    // Also write evidence for promoted/disliked signals
    if (action === 'promoted') {
      await this.recordTrackEvidence({
        user_id: userId,
        track_id: trackId,
        signal: 'add_to_favorites',
        profile: 'longterm',
        weight: 0.60,
      });
    } else if (action === 'disliked') {
      await this.recordTrackEvidence({
        user_id: userId,
        track_id: trackId,
        signal: 'hidden',
        profile: 'negative',
        weight: -0.60,
      });
    }
  }

  async getHistory(userId: string, limit = 50): Promise<HistoryEntry[]> {
    const res = await this.pgClient.query(
      `SELECT t.*, al.artwork_id, ph.played_at AS played_at, ph.completed AS completed, ph.id AS history_id, ph.batch_id AS batch_id
       FROM play_history ph
       JOIN tracks t ON t.id = ph.track_id
       LEFT JOIN albums al ON al.id = t.album_id
       WHERE ph.user_id = $1
       ORDER BY ph.played_at DESC
       LIMIT $2`,
      [userId, limit]
    );
    return res.rows as HistoryEntry[];
  }

  async createArtist(data: Artist): Promise<Artist> {
    const res = await this.pgClient.query(
      // `canonical_name` is NOT NULL with no default, so omitting it fails on a
      // fresh volume (the live DB's column predates the constraint). It holds the
      // DISPLAY name: the raw input, not normalize_artist()'s output, which
      // truncates on `/` and a standalone `x` ("AC/DC" -> "AC"). Matches the
      // convention in workers' scanner.service.resolveOrCreateArtist.
      `INSERT INTO artists (name, canonical_name, mbid, discogs_id, image_path)
       VALUES (normalize_artist($1), $2, $3, $4, $5) RETURNING *`,
      [data.name, data.name?.trim() || data.name, data.mbid, data.discogs_id, data.image_path]
    );
    return res.rows[0];
  }

  async updateArtist(id: string, data: Partial<Artist>): Promise<Artist> {
    const fields = allowedFields('artists', data);
    if (fields.length === 0) throw new Error('No fields to update');

    // Normalize name if it's being updated (the normalized_name generated column
    // handles it automatically, but we want the stored name itself to match).
    if (data.name !== undefined) {
      const norm = await this.pgClient.query('SELECT normalize_artist($1) AS n', [data.name]);
      data.name = norm.rows[0].n;
    }

    const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
    const values = fields.map(f => data[f as keyof Artist]);

    const res = await this.pgClient.query(
      `UPDATE artists SET ${setClause} WHERE id = $1 RETURNING *`,
      [id, ...values]
    );
    return res.rows[0];
  }

  async deleteArtist(id: string): Promise<void> {
    await this.pgClient.query('DELETE FROM artists WHERE id = $1', [id]);
  }

  async createAlbum(data: Album): Promise<Album> {
    const res = await this.pgClient.query(
      `INSERT INTO albums (artist_id, title, year, release_date, artwork_id)
       VALUES ($1, $2, $3, $4::date, $5) RETURNING *`,
      [data.artist_id, data.title, data.year, data.release_date ?? null, data.artwork_id]
    );
    return res.rows[0];
  }

  async updateAlbum(id: string, data: Partial<Album>): Promise<Album> {
    const fields = allowedFields('albums', data);
    if (fields.length === 0) throw new Error('No fields to update');
    
    const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
    const values = fields.map(f => (data as any)[f]);
    
    const res = await this.pgClient.query(
      `UPDATE albums SET ${setClause} WHERE id = $1 RETURNING *`,
      [id, ...values]
    );
    return res.rows[0];
  }

  async deleteAlbum(id: string): Promise<void> {
    await this.pgClient.query('DELETE FROM albums WHERE id = $1', [id]);
  }

  async getTrackById(id: string): Promise<Track | null> {
    const res = await this.pgClient.query(
      `SELECT t.*, al.artwork_id FROM tracks t
       LEFT JOIN albums al ON al.id = t.album_id
       WHERE t.id = $1`, [id]);
    return (res.rows[0] as Track) || null;
  }

  async getTrackLyrics(trackId: string): Promise<{ lyrics_text: string | null; synced_lyrics: unknown | null; provider: string | null } | null> {
    const res = await this.pgClient.query(
      'SELECT lyrics_text, synced_lyrics, provider FROM track_lyrics WHERE track_id = $1',
      [trackId]
    );
    return res.rows[0] ?? null;
  }

  async createTrack(data: Track): Promise<Track> {
    const res = await this.pgClient.query(
      `INSERT INTO tracks (path, hash, title, artist, album_id, duration, state, source_type) 
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8) RETURNING *`,
      [data.path, data.hash, data.title, data.artist, data.album_id, data.duration, data.state, data.source_type]
    );
    return res.rows[0];
  }

  async updateTrack(id: string, data: Partial<Track>): Promise<Track> {
    const fields = allowedFields('tracks', data);
    if (fields.length === 0) throw new Error('No fields to update');
    
    const setClause = fields.map((f, i) => `"${f}" = $${i + 2}`).join(', ');
    const values = fields.map(f => (data as any)[f]);
    
    const res = await this.pgClient.query(
      `UPDATE tracks SET ${setClause} WHERE id = $1 RETURNING *`,
      [id, ...values]
    );
    return res.rows[0];
  }

  async deleteTrack(id: string): Promise<void> {
    await this.pgClient.query('DELETE FROM tracks WHERE id = $1', [id]);
  }

  /**
   * Permanently delete a disliked track, skipping the grace period. Cascade-
   * deletes the DB record (which also removes the dislikes row), writes the
   * 'deleted_permanent' audit row, and hands the file itself to the worker.
   *
   * This method does NOT unlink. The backend mounts /music `:ro` on purpose —
   * nothing reachable from an HTTP request may write to the library — so the
   * unlink is queued as a pending_file_deletions row and performed by the
   * cleanup sweep in the worker, the one process with an rw mount, and only when
   * MUZICK_ALLOW_HARD_DELETE is on. Until then the row is visible evidence of an
   * orphaned file rather than a lost one.
   *
   * Ordering: everything durable commits in one transaction, and the
   * irreversible filesystem act happens strictly afterwards. The previous
   * implementation unlinked first, which on this deployment meant EROFS and a
   * 500 with the track still present.
   *
   * `filePath` is accepted for call-site compatibility but the path is re-read
   * from the row under the transaction, so the queued deletion can never target
   * a stale or caller-supplied path.
   */
  async permanentlyDeleteTrack(userId: string, trackId: string, _filePath?: string): Promise<void> {
    await this.withTransaction(async (client) => {
      const trackRes = await client.query(
        'SELECT path, title, artist FROM tracks WHERE id = $1 FOR UPDATE',
        [trackId]
      );
      const track = trackRes.rows[0] as
        | { path: string; title: string | null; artist: string | null }
        | undefined;
      if (!track) return; // Already gone; nothing to audit or unlink.

      // Denormalised identity: track_id is ON DELETE SET NULL, so without these
      // the surviving audit row would not say which track was destroyed.
      await client.query(
        `INSERT INTO feedback (user_id, track_id, action, track_path, track_title, track_artist)
         VALUES ($1, $2, 'deleted_permanent', $3, $4, $5)`,
        [userId, trackId, track.path, track.title, track.artist]
      );

      await client.query(
        `INSERT INTO pending_file_deletions (path, track_id, track_title, track_artist)
         VALUES ($1, $2, $3, $4)
         ON CONFLICT (path) DO UPDATE SET requested_at = NOW()`,
        [track.path, trackId, track.title, track.artist]
      );

      // CASCADE deletes dislikes, play_history, track_genre, etc.
      await client.query('DELETE FROM tracks WHERE id = $1', [trackId]);
    });

    // Strongest negative signal (spec §B.3). entity_id has no FK to tracks, so
    // the row survives the deletion. Outside the transaction: a failure here
    // must not resurrect an already-audited deletion.
    await this.recordEvidence({
      user_id: userId,
      entity_type: 'track',
      entity_id: trackId,
      signal: 'manual_deleted',
      profile: 'negative',
      weight: -0.90,
    });
  }

  /**
   * Return duplicate track groups, keyed by the given mode.
   *
   * `hash` (default) — groups of tracks with the same content hash (byte-
   * identical duplicates). Excludes placeholder-hash rows that were written
   * before proper hashing existed.
   *
   * `title-artist` — groups of tracks with the same normalised title AND
   * normalised artist across different albums (catches the same song appearing
   * in a compilation or reissue). Excludes same-hash groups since those are
   * already caught by the hash mode.
   *
   * Each group returned as `{ key, tracks }` where `key` is the hash (mode
   * `hash`) or `"title // artist"` (mode `title-artist`).
   */
  async getDuplicateGroups(
    mode: 'hash' | 'title-artist' = 'hash'
  ): Promise<{ key: string; tracks: Track[] }[]> {
    if (mode === 'title-artist') {
      const res = await this.pgClient.query<Track & { normalized_title: string; normalized_artist: string }>(
        `SELECT t.*, LOWER(TRIM(t.title)) AS normalized_title
           FROM tracks t
           JOIN (
             SELECT LOWER(TRIM(title)) AS nt, normalized_artist AS na
             FROM tracks
             WHERE state = 'LIBRARY'
               AND hash IS NOT NULL AND hash <> '' AND hash <> 'placeholder-hash'
               AND normalized_artist IS NOT NULL AND normalized_artist <> ''
             GROUP BY nt, na
             HAVING COUNT(*) > 1
           ) dup ON LOWER(TRIM(t.title)) = dup.nt AND t.normalized_artist = dup.na
           WHERE t.state = 'LIBRARY'
             AND t.hash IS NOT NULL AND t.hash <> '' AND t.hash <> 'placeholder-hash'
             AND t.normalized_artist IS NOT NULL AND t.normalized_artist <> ''
           ORDER BY dup.nt, dup.na, t.play_count DESC, t.last_played_at DESC NULLS LAST`
      );

      const groups = new Map<string, Track[]>();
      for (const row of res.rows) {
        const key = `${row.normalized_title} // ${row.normalized_artist}`;
        const list = groups.get(key) ?? [];
        list.push(row);
        groups.set(key, list);
      }

      // Exclude groups where all tracks have the same hash (hash dedup already
      // covers those).
      const result: { key: string; tracks: Track[] }[] = [];
      for (const [key, tracks] of groups) {
        const uniqueHashes = new Set(tracks.map((t) => t.hash));
        if (uniqueHashes.size <= 1) continue;
        result.push({ key, tracks });
      }
      return result;
    }

    // Default: hash-based dedup.
    const res = await this.pgClient.query<Track & { hash: string }>(
      `SELECT t.*
         FROM tracks t
         JOIN (
           SELECT hash FROM tracks
           WHERE hash IS NOT NULL AND hash <> '' AND hash <> 'placeholder-hash'
           GROUP BY hash HAVING COUNT(*) > 1
         ) dup ON dup.hash = t.hash
         ORDER BY t.hash, t.play_count DESC, t.last_played_at DESC NULLS LAST`
    );
    const groups = new Map<string, Track[]>();
    for (const row of res.rows) {
      const list = groups.get(row.hash) ?? [];
      list.push(row);
      groups.set(row.hash, list);
    }
    return [...groups.entries()].map(([key, tracks]) => ({ key, tracks }));
  }

  /**
   * Keep `keepId`, re-parent its play history / feedback onto it, delete the
   * remaining files from disk, then hard-delete the rest. All inside a
   * transaction so a partial failure leaves no orphans.
   *
   * Like permanentlyDeleteTrack(), this no longer unlinks: the loser files are
   * queued into pending_file_deletions inside the same transaction and removed
   * later by the worker's gated cleanup sweep. Previously the unlink ran BEFORE
   * the transaction and rethrew anything other than ENOENT, so on this
   * deployment (/music mounted `:ro` in the backend) every dedup merge failed
   * with EROFS before touching the DB at all.
   */
  async mergeDuplicates(keepId: string, deleteIds: string[]): Promise<void> {
    // 1. Fetch paths + identity for the tracks being deleted (before they go).
    const { rows: losers } = await this.pgClient.query<{
      id: string;
      path: string;
      title: string | null;
      artist: string | null;
    }>(
      `SELECT id, path, title, artist FROM tracks WHERE id = ANY($1::uuid[])`,
      [deleteIds]
    );

    // 2. DB transaction: queue the file removals, re-parent history, delete rows.
    await this.withTransaction(async (client) => {
      for (const row of losers) {
        await client.query(
          `INSERT INTO pending_file_deletions (path, track_id, track_title, track_artist)
           VALUES ($1, $2, $3, $4)
           ON CONFLICT (path) DO UPDATE SET requested_at = NOW()`,
          [row.path, row.id, row.title, row.artist]
        );
      }

      for (const id of deleteIds) {
        await client.query(
          `UPDATE play_history SET track_id = $1 WHERE track_id = $2`,
          [keepId, id]
        );
        await client.query(
          `UPDATE feedback SET track_id = $1 WHERE track_id = $2`,
          [keepId, id]
        );
        await client.query(`DELETE FROM tracks WHERE id = $1`, [id]);
      }
    });
  }

  /**
   * Advance a dislike to WARNED state: set warned_at and update state.
   * Called by the cleanup sweep worker after the grace period expires.
   */
  async markDislikeWarned(trackId: string): Promise<void> {
    await this.pgClient.query(
      `UPDATE dislikes SET warned_at = NOW(), state = 'WARNED' WHERE track_id = $1`,
      [trackId]
    );
  }

  // =========================================================================
  // v2 — System A: Knowledge Graph (probabilistic fusion)
  // =========================================================================

  /**
   * UPSERT a claim into the graph. Idempotent: same (subject, predicate, object,
   * source, user_id) refreshes last_reinforced_at without duplicating.
   */
  async upsertClaim(claim: {
    user_id?: string | null;
    subject_type: string;
    subject_id: string;
    predicate: string;
    object_type: string;
    object_id: string;
    source: string;
    confidence?: number;
    raw?: unknown;
  }, client?: Queryable): Promise<string> {
    const res = await (client ?? this.pgClient).query(
      `INSERT INTO claims (user_id, subject_type, subject_id, predicate, object_type, object_id, source, confidence, raw)
       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
       ON CONFLICT (subject_type, subject_id, predicate, object_type, object_id, source, user_id)
       DO UPDATE SET last_reinforced_at = NOW(), evidence_at = NOW(), confidence = $8, raw = COALESCE($9, claims.raw)
       RETURNING id`,
      [
        claim.user_id ?? null,
        claim.subject_type,
        claim.subject_id,
        claim.predicate,
        claim.object_type,
        claim.object_id,
        claim.source,
        claim.confidence ?? 1.0,
        claim.raw ? JSON.stringify(claim.raw) : null,
      ]
    );
    return res.rows[0].id as string;
  }

  /**
   * Batch UPSERT claims. Wraps multiple upsertClaim calls in a transaction.
   */
  async upsertClaims(claims: Parameters<DbService['upsertClaim']>[0][]): Promise<string[]> {
    return this.withTransaction(async (client) => {
      const ids: string[] = [];
      for (const claim of claims) {
        ids.push(await this.upsertClaim(claim, client));
      }
      return ids;
    });
  }

  /**
   * Get claims by subject (entity + predicate filter).
   */
  async getClaimsBySubject(
    subjectType: string,
    subjectId: string,
    predicate?: string,
    userId?: string
  ): Promise<Claim[]> {
    let sql = `SELECT * FROM claims WHERE subject_type = $1 AND subject_id = $2`;
    const params: unknown[] = [subjectType, subjectId];
    let idx = 3;

    if (predicate) {
      sql += ` AND predicate = $${idx}`;
      params.push(predicate);
      idx++;
    }
    if (userId) {
      sql += ` AND (user_id IS NULL OR user_id = $${idx})`;
      params.push(userId);
    }

    sql += ` ORDER BY last_reinforced_at DESC`;
    const res = await this.pgClient.query(sql, params);
    return res.rows as Claim[];
  }

  /**
   * Get fused value for a (subject, predicate, object) triple, optionally
   * scoped to a user (includes user-keyed claims).
   */
  async getFusedValue(
    subjectType: string,
    subjectId: string,
    predicate: string,
    objectType: string,
    objectId: string,
    userId?: string
  ): Promise<number> {
    let sql = `
      SELECT COALESCE(SUM(st.trust * c.confidence *
        GREATEST(0.1, 1.0 - EXTRACT(DAY FROM NOW() - c.last_reinforced_at) / 180.0)), 0) AS fused
      FROM claims c
      JOIN source_trust st ON st.key = c.source
      WHERE c.subject_type = $1 AND c.subject_id = $2
        AND c.predicate = $3
        AND c.object_type = $4 AND c.object_id = $5
    `;
    const params: unknown[] = [subjectType, subjectId, predicate, objectType, objectId];

    if (userId) {
      sql += ` AND (c.user_id IS NULL OR c.user_id = $6)`;
      params.push(userId);
    } else {
      sql += ` AND c.user_id IS NULL`;
    }

    const res = await this.pgClient.query(sql, params);
    return (res.rows[0]?.fused as number) ?? 0;
  }

  /**
   * Get fused artist credits for a track, returning the same shape as the old
   * attachArtists() for backward compatibility, but sourced from claim_fusion.
   */
  async getFusedTrackArtists(trackId: string, userId?: string): Promise<TrackArtist[]> {
    let sql = `
      SELECT a.id, a.name,
             CASE cf.predicate WHEN 'credited_main_on' THEN 'main' ELSE 'featured' END AS role,
             cf.fused_value AS confidence
      FROM claim_fusion cf
      JOIN artists a ON a.id = cf.object_id
      WHERE cf.subject_type = 'track' AND cf.subject_id = $1
        AND cf.predicate IN ('credited_main_on', 'featured_on')
        AND cf.object_type = 'artist'
    `;
    const params: unknown[] = [trackId];

    if (userId) {
      sql += ` AND cf.user_id IN ('00000000-0000-0000-0000-000000000000', $2) ORDER BY cf.fused_value DESC, cf.predicate`;
      params.push(userId);
    } else {
      sql += ` AND cf.user_id = '00000000-0000-0000-0000-000000000000' ORDER BY cf.fused_value DESC, cf.predicate`;
    }

    const res = await this.pgClient.query(sql, params);
    return res.rows as TrackArtist[];
  }

  // =========================================================================
  // v2 — System B: Listener Model
  // =========================================================================

  private beliefDimensionForSignal(signal: string): string {
    return signal === 'play_of_never_seen' ? 'novelty_tolerance' : 'affinity';
  }

  /**
   * Resolve the durable identities represented by a track. Artist credits use
   * the fusion-backed view (with the legacy table as a fallback during an
   * enrichment transition); genres and audio features are direct metadata.
   */
  private async getTrackBeliefTargets(trackId: string, client?: Queryable): Promise<Array<{
    entity_type: 'artist' | 'genre' | 'audio';
    entity_id: string;
    factor: number;
    context: Record<string, unknown>;
  }>> {
    const queryable = client ?? this.pgClient;
    const targets: Array<{
      entity_type: 'artist' | 'genre' | 'audio';
      entity_id: string;
      factor: number;
      context: Record<string, unknown>;
    }> = [];

    const identities = await queryable.query(
      `WITH artist_ids AS (
         SELECT artist_id FROM track_artists_v2 WHERE track_id = $1
         UNION
         SELECT artist_id FROM track_artists WHERE track_id = $1
       )
       SELECT 'artist' AS entity_type, artist_id AS entity_id FROM artist_ids
       UNION ALL
       SELECT 'genre' AS entity_type, genre_id AS entity_id
       FROM track_genre WHERE track_id = $1`,
      [trackId]
    );
    for (const row of identities.rows as Array<{ entity_type: 'artist' | 'genre'; entity_id: string }>) {
      targets.push({
        entity_type: row.entity_type,
        entity_id: row.entity_id,
        // An explicit favourite should be enough to form a usable comfort
        // artist belief; ordinary completed plays still accumulate gradually.
        factor: row.entity_type === 'artist' ? 0.90 : 0.45,
        context: { source_track_id: trackId, association: row.entity_type },
      });
    }

    const featuresRes = await queryable.query(
      `SELECT energy, bpm, valence
       FROM track_audio_features
       WHERE track_id = $1`,
      [trackId]
    );
    const features = featuresRes.rows[0] as { energy?: number | null; bpm?: number | null; valence?: number | null } | undefined;
    if (!features) return targets;

    const addAudioBucket = (dimension: 'energy' | 'bpm' | 'valence', bucket: string) => {
      targets.push({
        entity_type: 'audio',
        entity_id: bucket,
        factor: 0.30,
        context: { source_track_id: trackId, association: 'audio', dimension },
      });
    };
    if (typeof features.energy === 'number' && Number.isFinite(features.energy)) {
      addAudioBucket('energy', features.energy < 0.34
        ? AUDIO_PREFERENCE_BUCKETS.energy.low
        : features.energy < 0.67 ? AUDIO_PREFERENCE_BUCKETS.energy.medium : AUDIO_PREFERENCE_BUCKETS.energy.high);
    }
    if (typeof features.bpm === 'number' && Number.isFinite(features.bpm) && features.bpm > 0) {
      addAudioBucket('bpm', features.bpm < 90
        ? AUDIO_PREFERENCE_BUCKETS.bpm.slow
        : features.bpm <= 140 ? AUDIO_PREFERENCE_BUCKETS.bpm.medium : AUDIO_PREFERENCE_BUCKETS.bpm.fast);
    }
    if (typeof features.valence === 'number' && Number.isFinite(features.valence)) {
      addAudioBucket('valence', features.valence < 0.34
        ? AUDIO_PREFERENCE_BUCKETS.valence.low
        : features.valence < 0.67 ? AUDIO_PREFERENCE_BUCKETS.valence.neutral : AUDIO_PREFERENCE_BUCKETS.valence.high);
    }
    return targets;
  }

  /**
   * Append the track-level event, then project it onto the track's meaningful
   * shared identities. The original event remains the canonical audit record;
   * projected evidence makes artist/genre/audio affinity directly queryable by
   * Vibe generators. Callers pass their transaction client to keep the event
   * and every derived belief atomic.
   */
  async recordTrackEvidence(evidence: {
    user_id: string;
    track_id: string;
    signal: string;
    profile: string;
    weight: number;
    dimension?: string;
    context?: Record<string, unknown>;
  }, client?: Queryable): Promise<string> {
    const { track_id: trackId, ...event } = evidence;
    const id = await this.recordEvidence({
      ...event,
      entity_type: 'track',
      entity_id: trackId,
    }, client);
    const targets = await this.getTrackBeliefTargets(trackId, client);
    for (const target of targets) {
      await this.recordEvidence({
        user_id: event.user_id,
        entity_type: target.entity_type,
        entity_id: target.entity_id,
        signal: event.signal,
        profile: event.profile,
        weight: event.weight * target.factor,
        dimension: event.dimension,
        context: { ...event.context, ...target.context },
      }, client);
    }
    return id;
  }

  /**
   * Record evidence (append-only). Writes a signal into the evidence stream.
   */
  async recordEvidence(evidence: {
    user_id: string;
    entity_type: string;
    entity_id: string;
    signal: string;
    profile: string;
    weight: number;
    dimension?: string;
    context?: unknown;
  }, client?: Queryable): Promise<string> {
    const res = await (client ?? this.pgClient).query(
      `INSERT INTO evidence (user_id, entity_type, entity_id, signal, profile, weight, context)
       VALUES ($1, $2, $3, $4, $5, $6, $7) RETURNING id`,
      [
        evidence.user_id,
        evidence.entity_type,
        evidence.entity_id,
        evidence.signal,
        evidence.profile,
        evidence.weight,
        evidence.context ? JSON.stringify(evidence.context) : null,
      ]
    );
    const id = res.rows[0].id as string;

    // Derive the belief dimension from the signal. Per spec §B.3, every
    // signal feeds the 'affinity' dimension EXCEPT 'play_of_never_seen',
    // which feeds 'novelty_tolerance'. Each new evidence row must also
    // upsert the matching listener_belief (spec §B.4) — otherwise evidence
    // accumulates but beliefs never materialise.
    const dimension = evidence.dimension ?? this.beliefDimensionForSignal(evidence.signal);
    await this.updateListenerBelief({
      user_id: evidence.user_id,
      profile: evidence.profile,
      entity_type: evidence.entity_type,
      entity_id: evidence.entity_id,
      dimension,
      value_delta: evidence.weight,
      confidence_delta: 0.05,
    }, client);

    return id;
  }

  /**
   * Batch record evidence. All or nothing.
   */
  async recordEvidenceBatch(
    evidenceList: Parameters<DbService['recordEvidence']>[0][]
  ): Promise<string[]> {
    return this.withTransaction(async (client) => {
      const ids: string[] = [];
      for (const ev of evidenceList) {
        ids.push(await this.recordEvidence(ev, client));
      }
      return ids;
    });
  }

  /**
   * Rebuild only the derived shared-preference layer from durable local
   * interaction history. This intentionally does not append new evidence (the
   * evidence log is an audit stream) and does not replace track beliefs. It is
   * safe to run repeatedly after deploying propagation or after enrichment has
   * added artist/genre/audio metadata to old tracks.
   */
  async rebuildDerivedListenerBeliefs(userId: string): Promise<{ interactions: number; beliefs: number }> {
    return this.withTransaction(async (client) => {
      await client.query(
        `DELETE FROM listener_beliefs
         WHERE user_id = $1 AND entity_type IN ('artist', 'genre', 'audio')`,
        [userId]
      );

      const interactionRes = await client.query(
        `SELECT track_id, signal, profile, weight
         FROM (
           SELECT ph.track_id, ph.played_at AS occurred_at,
                  'playback_completed'::text AS signal,
                  'longterm'::text AS profile,
                  0.10::real AS weight
           FROM play_history ph
           WHERE ph.user_id = $1 AND ph.completed = true

           UNION ALL

           SELECT replay.track_id, replay.played_at AS occurred_at,
                  'replay_within_24h'::text AS signal,
                  'longterm'::text AS profile,
                  0.25::real AS weight
           FROM (
             SELECT track_id, played_at,
                    COUNT(*) OVER (
                      PARTITION BY track_id ORDER BY played_at
                      RANGE BETWEEN INTERVAL '24 hours' PRECEDING AND CURRENT ROW
                    ) AS recent_plays
             FROM play_history WHERE user_id = $1 AND completed = true
           ) replay
           WHERE replay.recent_plays > 1

           UNION ALL

           SELECT replay.track_id, replay.played_at AS occurred_at,
                  'replay_within_24h'::text AS signal,
                  'obsession'::text AS profile,
                  0.40::real AS weight
           FROM (
             SELECT track_id, played_at,
                    COUNT(*) OVER (
                      PARTITION BY track_id ORDER BY played_at
                      RANGE BETWEEN INTERVAL '24 hours' PRECEDING AND CURRENT ROW
                    ) AS recent_plays
             FROM play_history WHERE user_id = $1 AND completed = true
           ) replay
           WHERE replay.recent_plays > 1

           UNION ALL

           SELECT fav.track_id, fav.created_at AS occurred_at,
                  'add_to_favorites'::text AS signal,
                  'longterm'::text AS profile,
                  0.60::real AS weight
           FROM favorites fav WHERE fav.user_id = $1

           UNION ALL

           SELECT f.track_id, f.created_at AS occurred_at,
                  CASE f.action
                    WHEN 'promoted' THEN 'add_to_favorites'
                    WHEN 'disliked' THEN 'hidden'
                    WHEN 'skipped' THEN 'skip_quick'
                  END AS signal,
                  CASE WHEN f.action = 'promoted' THEN 'longterm' ELSE 'negative' END AS profile,
                  CASE f.action
                    WHEN 'promoted' THEN 0.60::real
                    WHEN 'disliked' THEN -0.60::real
                    WHEN 'skipped' THEN -0.20::real
                  END AS weight
           FROM feedback f
           WHERE f.user_id = $1
             AND f.track_id IS NOT NULL
             AND f.action IN ('promoted', 'disliked', 'skipped')
         ) interactions
         ORDER BY occurred_at ASC`,
        [userId]
      );

      let beliefs = 0;
      for (const interaction of interactionRes.rows as Array<{
        track_id: string;
        signal: string;
        profile: string;
        weight: number;
      }>) {
        const targets = await this.getTrackBeliefTargets(interaction.track_id, client);
        for (const target of targets) {
          await this.updateListenerBelief({
            user_id: userId,
            profile: interaction.profile,
            entity_type: target.entity_type,
            entity_id: target.entity_id,
            dimension: this.beliefDimensionForSignal(interaction.signal),
            value_delta: interaction.weight * target.factor,
            confidence_delta: 0.05,
          }, client);
          beliefs++;
        }
      }
      return { interactions: interactionRes.rows.length, beliefs };
    });
  }

  /**
   * Get listener beliefs for a user, optionally filtered by profile/entity.
   */
  async getListenerBeliefs(params: {
    userId: string;
    profile?: string;
    entityType?: string;
    entityId?: string;
    dimension?: string;
    limit?: number;
    orderBy?: 'value' | 'last_reinforced_at';
    order?: 'ASC' | 'DESC';
  }): Promise<ListenerBelief[]> {
    const {
      userId, profile, entityType, entityId, dimension,
      limit = 50, orderBy = 'value', order = 'DESC',
    } = params;

    let sql = `SELECT * FROM listener_beliefs WHERE user_id = $1`;
    const sqlParams: unknown[] = [userId];
    let idx = 2;

    if (profile) { sql += ` AND profile = $${idx}`; sqlParams.push(profile); idx++; }
    if (entityType) { sql += ` AND entity_type = $${idx}`; sqlParams.push(entityType); idx++; }
    if (entityId) { sql += ` AND entity_id = $${idx}`; sqlParams.push(entityId); idx++; }
    if (dimension) { sql += ` AND dimension = $${idx}`; sqlParams.push(dimension); idx++; }

    const validOrderBy = ['value', 'last_reinforced_at', 'confidence', 'evidence_count'];
    const sortCol = validOrderBy.includes(orderBy) ? orderBy : 'value';
    const sortOrder = order === 'ASC' ? 'ASC' : 'DESC';
    sql += ` ORDER BY ${sortCol} ${sortOrder} LIMIT $${idx}`;
    sqlParams.push(limit);

    const res = await this.pgClient.query(sql, sqlParams);
    return res.rows as ListenerBelief[];
  }

  /**
   * UPSERT a listener belief. Updates value, confidence, and evidence_count.
   * Implements the belief update formula from spec §B.4.
   */
  async updateListenerBelief(params: {
    user_id: string;
    profile: string;
    entity_type: string;
    entity_id: string;
    dimension: string;
    value_delta: number;
    confidence_delta?: number;
  }, client?: Queryable): Promise<void> {
    const { user_id, profile, entity_type, entity_id, dimension, value_delta, confidence_delta = 0.05 } = params;

    await (client ?? this.pgClient).query(
      `INSERT INTO listener_beliefs (user_id, profile, entity_type, entity_id, dimension, value, confidence, evidence_count, last_reinforced_at)
       VALUES ($1, $2, $3, $4, $5, $6, $7, 1, NOW())
       ON CONFLICT (user_id, profile, entity_type, entity_id, dimension)
       DO UPDATE SET
         value = GREATEST(-1.0, LEAST(1.0, listener_beliefs.value + $6 * (1.0 - listener_beliefs.confidence))),
         confidence = GREATEST(0, LEAST(1.0, listener_beliefs.confidence + $7)),
         evidence_count = listener_beliefs.evidence_count + 1,
         last_reinforced_at = NOW()`,
      [user_id, profile, entity_type, entity_id, dimension, value_delta, confidence_delta]
    );
  }

  // =========================================================================
  // v2 — System D: Session support
  // =========================================================================

  /**
   * Create a new session state row.
   */
  async createSessionState(
    userId: string,
    context?: string,
    stateVector?: Record<string, unknown>,
    sessionId?: string
  ): Promise<string> {
    const res = await this.pgClient.query(
      `INSERT INTO session_state (session_id, user_id, context, state_vector)
       VALUES (COALESCE($1::uuid, gen_random_uuid()), $2, $3, $4::jsonb)
       RETURNING session_id`,
      [sessionId ?? null, userId, context ?? null, stateVector ? JSON.stringify(stateVector) : '{}']
    );
    return res.rows[0].session_id as string;
  }

  /**
   * Get the most recent session state for a user.
   */
  async getLatestSessionState(userId: string): Promise<SessionState | null> {
    const res = await this.pgClient.query(
      `SELECT * FROM session_state WHERE user_id = $1 ORDER BY last_interaction DESC LIMIT 1`,
      [userId]
    );
    return (res.rows[0] as SessionState) || null;
  }

  /**
   * Create the authoritative Vibe v2 session record. This intentionally does
   * not create a legacy session_state row: callers can migrate to the durable
   * ledger without changing the existing v2 endpoint contract first.
   */
  async createVibeSession(params: {
    userId: string;
    policyVersion: string;
    seedTrackId?: string | null;
    context?: Record<string, unknown>;
    profile?: {
      goals: Record<string, unknown>;
      explorationCoefficient: number;
      discoveryRadius: number;
    };
  }): Promise<VibeSession> {
    return this.withTransaction(async (client) => {
      // Serialize starts for one listener even when there is no active row to
      // lock yet. The row lock below then safely replaces any prior session.
      await client.query(`SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, [params.userId]);
      // Lock every active session first. This makes concurrent starts converge
      // on one active durable session instead of creating overlapping streams.
      const active = await client.query(
        `SELECT id FROM vibe_sessions WHERE user_id = $1 AND status = 'active' FOR UPDATE`,
        [params.userId],
      );
      if (active.rows.length > 0) {
        const replaced = await client.query(
          `UPDATE vibe_sessions
           SET status = 'replaced', ended_at = NOW(), last_event_at = NOW()
           WHERE user_id = $1 AND status = 'active'
           RETURNING id`,
          [params.userId],
        );
        for (const session of replaced.rows as Array<{ id: string }>) {
          await client.query(
            `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload)
             VALUES ($1, $2, 'session_ended', NOW(), '{"reason":"replaced"}'::jsonb)`,
            [session.id, params.userId],
          );
        }
      }
      const res = await client.query(
        `WITH created AS (
           INSERT INTO vibe_sessions (user_id, status, seed_track_id, context, policy_version)
           VALUES ($1, 'active', $2, $3::jsonb, $4)
           RETURNING *
         ), profile AS (
           INSERT INTO vibe_session_profiles
             (session_id, user_id, goals, exploration_coefficient, discovery_radius)
           SELECT id, user_id, $5::jsonb, $6::real, $7::real FROM created
           ON CONFLICT (session_id) DO NOTHING
         )
         SELECT * FROM created`,
        [
          params.userId,
          params.seedTrackId ?? null,
          JSON.stringify(params.context ? { ...params.context } : {}),
          params.policyVersion,
          JSON.stringify(params.profile?.goals ?? { type: 'discovery', target: 1, progress: 0 }),
          params.profile?.explorationCoefficient ?? 0.3,
          params.profile?.discoveryRadius ?? 0.38,
        ],
      );
      return res.rows[0] as VibeSession;
    });
  }

  /** Fetch a Vibe session only when it belongs to the requesting user. */
  async getVibeSession(sessionId: string, userId: string): Promise<VibeSession | null> {
    const res = await this.pgClient.query(
      'SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2',
      [sessionId, userId]
    );
    return (res.rows[0] as VibeSession) ?? null;
  }

  /** Recent session shapes, excluding the active session. The fingerprint is
   * intentionally aggregate-only and is used as a soft planning penalty. */
  async getRecentVibeSessionFingerprints(userId: string, sessionId: string, limit = 8): Promise<Record<string, unknown>[]> {
    const res = await this.pgClient.query(
      `SELECT p.fingerprint
       FROM vibe_session_profiles p
       JOIN vibe_sessions s ON s.id = p.session_id
       WHERE p.user_id = $1 AND p.session_id <> $2::uuid
         AND s.status IN ('ended', 'expired', 'replaced')
         AND p.fingerprint <> '{}'::jsonb
       ORDER BY p.updated_at DESC
       LIMIT $3`,
      [userId, sessionId, limit],
    );
    return res.rows.map((row: { fingerprint: Record<string, unknown> }) => row.fingerprint ?? {});
  }

  async getVibeSessionProfile(sessionId: string, userId: string): Promise<VibeSessionProfile | null> {
    const res = await this.pgClient.query(
      `SELECT p.* FROM vibe_session_profiles p
       JOIN vibe_sessions s ON s.id = p.session_id
       WHERE p.session_id = $1 AND s.user_id = $2`,
      [sessionId, userId],
    );
    return (res.rows[0] as VibeSessionProfile | undefined) ?? null;
  }

  /**
   * Project unknown-track feedback into the session exploration controls.
   * This deliberately runs behind its own projection marker: the immutable
   * event has already committed, so retries after a transient failure are
   * safe and converge on one evidence row and one coefficient adjustment.
   */
  async projectVibeSessionFeedback(event: VibeEvent): Promise<void> {
    if (!event.track_id || !['completed', 'skipped', 'kept'].includes(event.type)) return;
    const trackId = event.track_id;
    await this.withTransaction(async client => {
      // Old active/resumable sessions predate vibe_session_profiles. Create a
      // neutral profile before claiming the exactly-once marker: otherwise the
      // marker could permanently consume a feedback event without adapting its
      // session. Existing goals are deliberately never overwritten here.
      await client.query(
        `INSERT INTO vibe_session_profiles (session_id, user_id)
         VALUES ($1, $2)
         ON CONFLICT (session_id) DO NOTHING`,
        [event.session_id, event.user_id],
      );
      const marker = await client.query(
        `INSERT INTO vibe_session_feedback_projections (event_id)
         VALUES ($1) ON CONFLICT (event_id) DO NOTHING RETURNING event_id`,
        [event.id],
      );
      if (!marker.rows[0]) return;

      const sessionContext = await client.query(
        'SELECT context FROM vibe_sessions WHERE id = $1 AND user_id = $2',
        [event.session_id, event.user_id],
      );
      const context = (sessionContext.rows[0]?.context ?? {}) as Record<string, unknown>;
      const contextDimension = calendarContextKey(context);
      if (contextDimension) {
        const contextWeight = event.type === 'skipped' ? -0.16 : event.type === 'completed' ? 0.08 : 0.04;
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: trackId,
          signal: event.type === 'skipped' ? 'skip_quick' : event.type === 'kept' ? 'kept' : 'playback_completed',
          profile: 'contextual',
          weight: contextWeight,
          dimension: contextDimension,
          context: { vibe_event_id: event.id, session_id: event.session_id, calendar: context },
        }, client);
      }

      // This query occurs before a completed event's play_history projection
      // can be considered. Favourites and prior evidence count as familiarity
      // too, avoiding a false “new discovery” on a locally known track.
      const familiarity = await client.query(
        `SELECT (
           EXISTS (SELECT 1 FROM play_history WHERE user_id = $1 AND track_id = $2 AND completed = true AND played_at < $3::timestamptz)
           OR EXISTS (SELECT 1 FROM favorites WHERE user_id = $1 AND track_id = $2)
           OR EXISTS (SELECT 1 FROM evidence WHERE user_id = $1 AND entity_type = 'track' AND entity_id = $2 AND created_at < $3::timestamptz)
         ) AS familiar`,
        [event.user_id, trackId, event.occurred_at],
      );
      const familiar = Boolean(familiarity.rows[0]?.familiar);

      const delta = event.type === 'skipped' ? -0.08 : event.type === 'completed' ? 0.06 : 0.03;
      const signal = event.type === 'skipped' ? 'skip_quick' : 'play_of_never_seen';
      const weight = event.type === 'skipped' ? -0.05 : delta;
      if (!familiar) {
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: trackId,
          signal,
          profile: event.type === 'skipped' ? 'negative' : 'discovery',
          weight,
          context: { vibe_event_id: event.id, session_id: event.session_id, unfamiliar: true },
        }, client);
      }

      const profile = await client.query(
        `UPDATE vibe_session_profiles
         SET exploration_coefficient = GREATEST(0, LEAST(1, exploration_coefficient + CASE WHEN $4 THEN 0::real ELSE $3::real END)),
             discovery_radius = GREATEST(0.15, LEAST(0.9, 0.2 + (GREATEST(0, LEAST(1, exploration_coefficient + CASE WHEN $4 THEN 0::real ELSE $3::real END)) * 0.65))),
             goals = CASE
               WHEN goals->>'type' = 'familiar' AND $4 AND $5
                 THEN jsonb_set(goals, '{progress}', to_jsonb(LEAST(COALESCE((goals->>'progress')::int, 0) + 1, COALESCE((goals->>'target')::int, 1))))
               WHEN goals->>'type' IN ('discovery', 'surprise', 'artist_introduction') AND NOT $4 AND $3::real > 0
                 THEN jsonb_set(goals, '{progress}', to_jsonb(LEAST(COALESCE((goals->>'progress')::int, 0) + 1, COALESCE((goals->>'target')::int, 1))))
               ELSE goals END,
             updated_at = NOW()
         WHERE session_id = $1 AND user_id = $2
         RETURNING exploration_coefficient, discovery_radius, goals`,
        [event.session_id, event.user_id, delta, familiar, event.type === 'completed' || event.type === 'kept'],
      );
      const row = profile.rows[0] as Pick<VibeSessionProfile, 'exploration_coefficient' | 'discovery_radius' | 'goals'> | undefined;
      if (row) {
        await client.query(
          `UPDATE session_state
           SET state_vector = state_vector || jsonb_build_object(
                 'explorationCoefficient', $3::real,
                 'discoveryRadius', $4::real,
                 'sessionGoal', $5::jsonb
               ), last_interaction = NOW()
           WHERE session_id = $1 AND user_id = $2`,
          [event.session_id, event.user_id, row.exploration_coefficient, row.discovery_radius, JSON.stringify(row.goals)],
        );
      }
    });
  }

  /**
   * End (or expire/replace) a session without changing its original end time
   * when a client retries the same request.
   */
  async endVibeSession(
    sessionId: string,
    userId: string,
    status: Extract<VibeSessionStatus, 'ended' | 'expired' | 'replaced'> = 'ended'
  ): Promise<VibeSession | null> {
    const res = await this.pgClient.query(
      `UPDATE vibe_sessions
       SET status = CASE WHEN ended_at IS NULL THEN $3 ELSE status END,
           ended_at = COALESCE(ended_at, NOW()),
           last_event_at = CASE WHEN ended_at IS NULL THEN NOW() ELSE last_event_at END
       WHERE id = $1 AND user_id = $2
       RETURNING *`,
      [sessionId, userId, status]
    );
    return (res.rows[0] as VibeSession) ?? null;
  }

  /**
   * Resume an owned paused/active session exactly once.  The row lock makes
   * the transition and its ledger entry inseparable and prevents retries from
   * manufacturing a stream of session_resumed events.
   */
  async resumeVibeSession(sessionId: string, userId: string): Promise<{ session: VibeSession; resumed: boolean }> {
    return this.withTransaction(async (client) => {
      const result = await client.query(
        `SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`,
        [sessionId, userId]
      );
      const session = result.rows[0] as VibeSession | undefined;
      if (!session) throw new Error('Vibe session was not found or is not owned by this user');
      if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') {
        throw new Error(`Cannot resume ${session.status} Vibe session`);
      }

      const prior = await client.query(
        `SELECT 1 FROM vibe_events WHERE session_id = $1 AND type = 'session_resumed' LIMIT 1`,
        [sessionId]
      );
      if (prior.rowCount) return { session, resumed: false };

      const updated = await client.query(
        `UPDATE vibe_sessions SET status = 'active', ended_at = NULL, last_event_at = NOW()
         WHERE id = $1 RETURNING *`,
        [sessionId]
      );
      const resumed = updated.rows[0] as VibeSession;
      await client.query(
        `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload)
         VALUES ($1, $2, 'session_resumed', NOW(), '{}'::jsonb)`,
        [sessionId, userId]
      );
      return { session: resumed, resumed: true };
    });
  }

  /** End a session and append its terminal event under one session-row lock. */
  async endVibeSessionWithEvent(sessionId: string, userId: string): Promise<{ session: VibeSession; ended: boolean }> {
    return this.withTransaction(async (client) => {
      const result = await client.query(
        `SELECT * FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`,
        [sessionId, userId]
      );
      const session = result.rows[0] as VibeSession | undefined;
      if (!session) throw new Error('Vibe session was not found or is not owned by this user');
      if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') {
        return { session, ended: false };
      }
      await client.query(
        `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload)
         VALUES ($1, $2, 'session_ended', NOW(), '{}'::jsonb)`,
        [sessionId, userId]
      );
      const updated = await client.query(
        `UPDATE vibe_sessions SET status = 'ended', ended_at = NOW(), last_event_at = NOW()
         WHERE id = $1 RETURNING *`,
        [sessionId]
      );
      return { session: updated.rows[0] as VibeSession, ended: true };
    });
  }

  /**
   * Append an immutable Vibe event. A supplied clientEventId is idempotent per
   * session: a retry returns the original event and does not advance the
   * session timestamp a second time. A missing id deliberately means a new,
   * server-originated event.
   */
  async recordVibeEvent(params: {
    sessionId: string;
    userId: string;
    type: string;
    clientEventId?: string | null;
    trackId?: string | null;
    occurredAt?: Date;
    positionMs?: number | null;
    durationMs?: number | null;
    payload?: Record<string, unknown>;
  }): Promise<RecordedVibeEvent> {
    return this.withTransaction(async (client) => {
      // A session-row lock serializes both event writes and terminal state
      // transitions. In particular, it avoids the READ COMMITTED CTE snapshot
      // race where ON CONFLICT observes a concurrent event but a later CTE
      // cannot yet read it. The duplicate lookup happens after the lock, so an
      // idempotent retry remains valid even after the session has ended.
      const sessionRes = await client.query(
        `SELECT id, status
         FROM vibe_sessions
         WHERE id = $1 AND user_id = $2
         FOR UPDATE`,
        [params.sessionId, params.userId]
      );
      const session = sessionRes.rows[0] as Pick<VibeSession, 'id' | 'status'> | undefined;
      if (!session) {
        throw new Error('Vibe session was not found or is not owned by this user');
      }

      if (params.clientEventId) {
        const existingRes = await client.query(
          `SELECT *
           FROM vibe_events
           WHERE session_id = $1 AND client_event_id = $2::uuid`,
          [params.sessionId, params.clientEventId]
        );
        const existing = existingRes.rows[0] as VibeEvent | undefined;
        if (existing) {
          await this.projectVibeFeedback(existing, client);
          return { event: existing, inserted: false };
        }
      }

      if (session.status === 'ended' || session.status === 'expired' || session.status === 'replaced') {
        throw new Error(`Cannot record a new event for ${session.status} Vibe session`);
      }

      const occurredAt = params.occurredAt?.toISOString() ?? null;
      const insertRes = await client.query(
        `INSERT INTO vibe_events
           (client_event_id, session_id, user_id, track_id, type, occurred_at, position_ms, duration_ms, payload)
         VALUES ($1::uuid, $2, $3, $4::uuid, $5, COALESCE($6::timestamptz, NOW()), $7, $8, $9::jsonb)
         RETURNING *`,
        [
          params.clientEventId ?? null,
          params.sessionId,
          params.userId,
          params.trackId ?? null,
          params.type,
          occurredAt,
          params.positionMs ?? null,
          params.durationMs ?? null,
          JSON.stringify(params.payload ?? {}),
        ]
      );
      const event = insertRes.rows[0] as VibeEvent | undefined;
      if (!event) {
        throw new Error('Vibe event could not be recorded');
      }

      await this.projectVibeFeedback(event, client);
      await client.query(
        `UPDATE vibe_sessions
         SET last_event_at = GREATEST(last_event_at, $2::timestamptz)
         WHERE id = $1`,
        [params.sessionId, event.occurred_at]
      );
      return { event, inserted: true };
    });
  }

  /**
   * Materialize Vibe feedback into the listener inputs used by the incumbent
   * director. The projection marker and every write share the event's
   * transaction, so a retry either sees the completed projection or performs
   * it once; it can never double-count a completed/skip/dislike/kept signal.
   */
  private async projectVibeFeedback(event: VibeEvent, client: PoolClient): Promise<void> {
    if (!event.track_id || !['completed', 'skipped', 'disliked', 'kept'].includes(event.type)) return;
    const projection = await client.query(
      `INSERT INTO vibe_event_projections (event_id)
       VALUES ($1)
       ON CONFLICT (event_id) DO NOTHING
       RETURNING event_id`,
      [event.id],
    );
    if (!projection.rows[0]) return;

    const occurredAt = event.occurred_at?.toISOString?.() ?? new Date().toISOString();
    switch (event.type) {
      case 'completed':
        // The Vibe player is the one caller that knows how much was actually
        // heard. position_ms is where playback reached; duration_ms is the
        // fallback for a client that only reported the track length.
        await client.query(
          `INSERT INTO play_history (
             user_id, track_id, completed, played_at, listened_ms, track_title, track_artist
           )
           SELECT $1::uuid, $2::uuid, true, $3::timestamptz, $4::int, t.title, t.artist
             FROM (SELECT 1) AS always
             LEFT JOIN tracks t ON t.id = $2::uuid`,
          [event.user_id, event.track_id, occurredAt, event.position_ms ?? event.duration_ms ?? null],
        );
        await client.query(
          `UPDATE tracks
           SET play_count = play_count + 1, last_played_at = $2::timestamptz
           WHERE id = $1`,
          [event.track_id, occurredAt],
        );
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: event.track_id,
          signal: 'playback_completed',
          profile: 'longterm',
          weight: 0.10,
          context: { vibe_event_id: event.id, session_id: event.session_id },
        }, client);
        break;
      case 'skipped':
        await client.query('UPDATE tracks SET skip_count = skip_count + 1 WHERE id = $1', [event.track_id]);
        await client.query(
          `INSERT INTO feedback (user_id, track_id, action, created_at)
           VALUES ($1, $2, 'skipped', $3::timestamptz)`,
          [event.user_id, event.track_id, occurredAt],
        );
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: event.track_id,
          signal: 'skip_quick',
          profile: 'negative',
          weight: -0.20,
          context: { vibe_event_id: event.id, session_id: event.session_id },
        }, client);
        break;
      case 'disliked':
        await client.query('UPDATE tracks SET dislike_count = dislike_count + 1 WHERE id = $1', [event.track_id]);
        await client.query(
          `INSERT INTO feedback (user_id, track_id, action, created_at)
           VALUES ($1, $2, 'disliked', $3::timestamptz)`,
          [event.user_id, event.track_id, occurredAt],
        );
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: event.track_id,
          signal: 'hidden',
          profile: 'negative',
          weight: -0.60,
          context: { vibe_event_id: event.id, session_id: event.session_id },
        }, client);
        break;
      case 'kept':
        await this.recordTrackEvidence({
          user_id: event.user_id,
          track_id: event.track_id,
          signal: 'kept',
          profile: 'longterm',
          weight: 0.05,
          context: { vibe_event_id: event.id, session_id: event.session_id },
        }, client);
        break;
    }
  }

  /**
   * Persist one complete revision of a session plan atomically. The caller
   * supplies the monotonically increasing version; session-level scheduling
   * will own version allocation when the director is migrated to this ledger.
   */
  async publishVibePlan(params: {
    sessionId: string;
    userId: string;
    /** Supplying one is for the first revision; otherwise allocate the next. */
    version?: number;
    reason: string;
    stateSnapshot: Record<string, unknown>;
    objectiveSnapshot: Record<string, unknown>;
    items: Array<Omit<VibePlanItem, 'plan_version_id'>>;
  }): Promise<VibePlan> {
    return this.withTransaction(async (client) => {
      // The session lock is also the concurrency boundary for starts/ends and
      // plan revisions. In particular, a slow initial planner cannot publish
      // into a session a newer start has already replaced.
      const sessionRes = await client.query(
        `SELECT id, status FROM vibe_sessions
         WHERE id = $1 AND user_id = $2
         FOR UPDATE`,
        [params.sessionId, params.userId],
      );
      const session = sessionRes.rows[0] as Pick<VibeSession, 'id' | 'status'> | undefined;
      if (!session) throw new Error('Vibe session was not found or is not owned by this user');
      if (session.status !== 'active') {
        throw new Error(`Cannot publish a plan for ${session.status} Vibe session`);
      }
      const version = params.version ?? Number((await client.query(
        `SELECT COALESCE(MAX(version), 0) + 1 AS version
         FROM vibe_plan_versions WHERE session_id = $1`,
        [params.sessionId],
      )).rows[0].version);
      const header = await client.query(
        `INSERT INTO vibe_plan_versions
           (session_id, version, reason, state_snapshot, objective_snapshot)
         VALUES ($1, $2, $3, $4::jsonb, $5::jsonb)
         RETURNING *`,
        [
          params.sessionId,
          version,
          params.reason,
          JSON.stringify(params.stateSnapshot),
          JSON.stringify(params.objectiveSnapshot),
        ],
      );
      const planVersion = header.rows[0] as VibePlan | undefined;
      if (!planVersion) throw new Error('Vibe plan could not be published');
      for (const item of params.items) {
        await client.query(
          `INSERT INTO vibe_plan_items
             (plan_version_id, ordinal, track_id, slot_role, candidate_source, score, score_breakdown, explanation, committed)
           VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9)`,
          [
            planVersion.id, item.ordinal, item.track_id, item.slot_role,
            item.candidate_source, item.score, JSON.stringify(item.score_breakdown),
            JSON.stringify(item.explanation), item.committed,
          ],
        );
      }
      // Header/items and this ledger event deliberately commit together. A
      // client retry can therefore find either neither or the same canonical
      // revision; it can never observe a published event without its plan.
      await client.query(
        `INSERT INTO vibe_events (session_id, user_id, type, occurred_at, payload)
         VALUES ($1, $2, 'plan_published', NOW(), $3::jsonb)`,
        [params.sessionId, params.userId, JSON.stringify({
          planVersion: planVersion.version,
          planVersionId: planVersion.id,
          reason: params.reason,
          itemCount: params.items.length,
          feedbackEventId: params.objectiveSnapshot.feedbackEventId ?? null,
        })],
      );
      // Store a compact session shape rather than a replayable queue. It is
      // overwritten on each revision so a recently adapted session represents
      // its current direction when tomorrow's session asks for freshness.
      await client.query(
        `WITH selected AS (
           SELECT i.track_id, i.candidate_source
           FROM vibe_plan_items i WHERE i.plan_version_id = $1
         ), artists AS (
           SELECT DISTINCT ta.artist_id::text AS value FROM selected s
           JOIN track_artists_v2 ta ON ta.track_id = s.track_id AND ta.role = 'main'
         ), genres AS (
           SELECT DISTINCT tg.genre_id::text AS value FROM selected s
           JOIN track_genre tg ON tg.track_id = s.track_id
         ), sources AS (
           SELECT DISTINCT candidate_source AS value FROM selected
         )
         UPDATE vibe_session_profiles
         SET fingerprint = jsonb_build_object(
               'artists', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM artists), '[]'::jsonb),
               'genres', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM genres), '[]'::jsonb),
               'sources', COALESCE((SELECT jsonb_agg(value ORDER BY value) FROM sources), '[]'::jsonb),
               'context', (SELECT context FROM vibe_sessions WHERE id = $2)
             ), updated_at = NOW()
         WHERE session_id = $2 AND user_id = $3`,
        [planVersion.id, params.sessionId, params.userId],
      );
      await client.query(
        `UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`,
        [params.sessionId],
      );
      return {
        ...planVersion,
        items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })),
      };
    });
  }

  async persistVibePlan(params: {
    sessionId: string;
    userId: string;
    version: number;
    reason: string;
    stateSnapshot: Record<string, unknown>;
    objectiveSnapshot: Record<string, unknown>;
    items: Array<Omit<VibePlanItem, 'plan_version_id'>>;
  }): Promise<VibePlan> {
    return this.withTransaction(async (client) => {
      const header = await client.query(
        `INSERT INTO vibe_plan_versions
           (session_id, version, reason, state_snapshot, objective_snapshot)
         SELECT s.id, $3, $4, $5::jsonb, $6::jsonb
         FROM vibe_sessions s
         WHERE s.id = $1 AND s.user_id = $2
         RETURNING *`,
        [
          params.sessionId,
          params.userId,
          params.version,
          params.reason,
          JSON.stringify(params.stateSnapshot),
          JSON.stringify(params.objectiveSnapshot),
        ]
      );
      const planVersion = header.rows[0] as VibePlan | undefined;
      if (!planVersion) {
        throw new Error('Vibe session was not found or is not owned by this user');
      }

      for (const item of params.items) {
        await client.query(
          `INSERT INTO vibe_plan_items
             (plan_version_id, ordinal, track_id, slot_role, candidate_source, score, score_breakdown, explanation, committed)
           VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9)`,
          [
            planVersion.id,
            item.ordinal,
            item.track_id,
            item.slot_role,
            item.candidate_source,
            item.score,
            JSON.stringify(item.score_breakdown),
            JSON.stringify(item.explanation),
            item.committed,
          ]
        );
      }

      return { ...planVersion, items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })) };
    });
  }

  /** Allocate and persist the next immutable revision while holding the session lock. */
  async persistNextVibePlan(params: Omit<Parameters<DbService['persistVibePlan']>[0], 'version'>): Promise<VibePlan> {
    return this.withTransaction(async (client) => {
      const session = await client.query(
        `SELECT id, status FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`,
        [params.sessionId, params.userId]
      );
      if (!session.rows[0]) throw new Error('Vibe session was not found or is not owned by this user');
      if ((session.rows[0] as Pick<VibeSession, 'status'>).status !== 'active') {
        throw new Error(`Cannot record a new event for ${(session.rows[0] as Pick<VibeSession, 'status'>).status} Vibe session`);
      }
      const versionResult = await client.query(
        `SELECT COALESCE(MAX(version), 0) + 1 AS version FROM vibe_plan_versions WHERE session_id = $1`,
        [params.sessionId]
      );
      const version = Number(versionResult.rows[0].version);
      const header = await client.query(
        `INSERT INTO vibe_plan_versions
           (session_id, version, reason, state_snapshot, objective_snapshot)
         VALUES ($1, $2, $3, $4::jsonb, $5::jsonb) RETURNING *`,
        [params.sessionId, version, params.reason, JSON.stringify(params.stateSnapshot), JSON.stringify(params.objectiveSnapshot)]
      );
      const planVersion = header.rows[0] as VibePlan;
      for (const item of params.items) {
        await client.query(
          `INSERT INTO vibe_plan_items
             (plan_version_id, ordinal, track_id, slot_role, candidate_source, score, score_breakdown, explanation, committed)
           VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8::jsonb, $9)`,
          [planVersion.id, item.ordinal, item.track_id, item.slot_role, item.candidate_source,
            item.score, JSON.stringify(item.score_breakdown), JSON.stringify(item.explanation), item.committed]
        );
      }
      return { ...planVersion, items: params.items.map((item) => ({ ...item, plan_version_id: planVersion.id })) };
    });
  }

  /**
   * Atomically commit one item from the latest plan. A version-aware request
   * acts as an idempotency key: retrying the same expected version receives
   * the original item, while a replaced plan is returned as stale without
   * committing any old item. Calls without an expected version preserve the
   * original legacy cursor behaviour.
   */
  async serveNextVibePlanItem(
    sessionId: string,
    userId: string,
    expectedPlanVersion?: number,
  ): Promise<{ item: VibePlanItem | null; stale: boolean }> {
    return this.withTransaction(async (client) => {
      const session = await client.query(
        `SELECT status FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`,
        [sessionId, userId]
      );
      const row = session.rows[0] as Pick<VibeSession, 'status'> | undefined;
      if (!row) throw new Error('Vibe session was not found or is not owned by this user');
      if (row.status !== 'active') throw new Error(`Cannot record a new event for ${row.status} Vibe session`);
      const latest = await client.query(
        `SELECT id, version FROM vibe_plan_versions WHERE session_id = $1 ORDER BY version DESC LIMIT 1 FOR UPDATE`,
        [sessionId],
      );
      const plan = latest.rows[0] as Pick<VibePlan, 'id' | 'version'> | undefined;
      if (!plan) return { item: null, stale: false };
      if (expectedPlanVersion !== undefined && expectedPlanVersion !== plan.version) {
        return { item: null, stale: true };
      }

      if (expectedPlanVersion !== undefined) {
        const prior = await client.query(
          `SELECT i.*
           FROM vibe_events e
           JOIN vibe_plan_items i
             ON i.plan_version_id = (e.payload->>'planVersionId')::uuid
            AND i.ordinal = (e.payload->>'ordinal')::integer
           WHERE e.session_id = $1
             AND e.type = 'track_served'
             AND e.payload->>'planVersion' = $2::text
           ORDER BY e.occurred_at ASC
           LIMIT 1`,
          [sessionId, expectedPlanVersion],
        );
        const servedPreviously = prior.rows[0] as VibePlanItem | undefined;
        if (servedPreviously) return { item: servedPreviously, stale: false };
      }
      const item = await client.query(
        `WITH next_item AS (
           SELECT i.plan_version_id, i.ordinal FROM vibe_plan_items i
           WHERE i.plan_version_id = $1 AND NOT i.committed ORDER BY i.ordinal ASC LIMIT 1 FOR UPDATE
         )
         UPDATE vibe_plan_items i SET committed = true
         FROM next_item n WHERE i.plan_version_id = n.plan_version_id AND i.ordinal = n.ordinal
         RETURNING i.*`,
        [plan.id]
      );
      const served = item.rows[0] as VibePlanItem | undefined;
      if (!served) return { item: null, stale: false };
      await client.query(
        `INSERT INTO vibe_events (session_id, user_id, track_id, type, occurred_at, payload)
         VALUES ($1, $2, $3, 'track_served', NOW(), $4::jsonb)`,
        [sessionId, userId, served.track_id, JSON.stringify({
          planVersion: plan.version,
          planVersionId: served.plan_version_id,
          ordinal: served.ordinal,
        })]
      );
      await client.query(`UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`, [sessionId]);
      return { item: served, stale: false };
    });
  }

  /**
   * Advance an already-served item which the player could not resolve (for
   * example, its file was hidden after the revision was published). Unlike a
   * normal version-aware /next retry, this has a distinct client event id and
   * therefore intentionally moves beyond the item previously served for that
   * revision. The event records both the rejected item and the replacement so
   * a lost response can be retried without consuming another plan item.
   */
  async advancePastUnplayableVibePlanItem(
    sessionId: string,
    userId: string,
    input: {
      expectedPlanVersion: number;
      planVersionId: string;
      ordinal: number;
      trackId: string;
      eventId: string;
    },
  ): Promise<{ item: VibePlanItem | null; stale: boolean }> {
    return this.withTransaction(async (client) => {
      const session = await client.query(
        `SELECT status FROM vibe_sessions WHERE id = $1 AND user_id = $2 FOR UPDATE`,
        [sessionId, userId],
      );
      const sessionRow = session.rows[0] as Pick<VibeSession, 'status'> | undefined;
      if (!sessionRow) throw new Error('Vibe session was not found or is not owned by this user');

      const latest = await client.query(
        `SELECT id, version FROM vibe_plan_versions WHERE session_id = $1 ORDER BY version DESC LIMIT 1 FOR UPDATE`,
        [sessionId],
      );
      const plan = latest.rows[0] as Pick<VibePlan, 'id' | 'version'> | undefined;
      if (!plan) return { item: null, stale: false };
      if (plan.version !== input.expectedPlanVersion || plan.id !== input.planVersionId) {
        return { item: null, stale: true };
      }

      // Idempotency is scoped to this explicit advancement operation, rather
      // than overloading the version-aware /next retry which must keep
      // returning the originally served item.
      const prior = await client.query(
        `SELECT type, payload
         FROM vibe_events
         WHERE session_id = $1 AND client_event_id = $2::uuid`,
        [sessionId, input.eventId],
      );
      const priorEvent = prior.rows[0] as Pick<VibeEvent, 'type' | 'payload'> | undefined;
      if (priorEvent && priorEvent.type !== 'playback_error') {
        throw new Error('Vibe client event id was already used for a different event');
      }
      const priorPayload = priorEvent?.payload as Record<string, unknown> | undefined;
      if (priorPayload) {
        if (priorPayload.planVersionId !== input.planVersionId
          || priorPayload.ordinal !== input.ordinal
          || priorPayload.trackId !== input.trackId) {
          throw new Error('Vibe playback-error event does not match the served plan item');
        }
        const advancedTo = priorPayload.advancedTo as { planVersionId?: unknown; ordinal?: unknown } | null | undefined;
        if (!advancedTo || typeof advancedTo.planVersionId !== 'string' || !Number.isInteger(advancedTo.ordinal)) {
          return { item: null, stale: false };
        }
        const replacement = await client.query(
          `SELECT * FROM vibe_plan_items WHERE plan_version_id = $1 AND ordinal = $2`,
          [advancedTo.planVersionId, advancedTo.ordinal],
        );
        return { item: (replacement.rows[0] as VibePlanItem | undefined) ?? null, stale: false };
      }

      if (sessionRow.status !== 'active') {
        throw new Error(`Cannot record a new event for ${sessionRow.status} Vibe session`);
      }

      // A client may advance only the cursor it was just served. Checking for
      // any historical serve event would let an old version-aware /next
      // response consume whichever future item happens to be uncommitted.
      const current = await client.query(
        `SELECT i.track_id, i.ordinal
         FROM vibe_events e
         JOIN vibe_plan_items i
           ON i.plan_version_id = $3::uuid
          AND i.ordinal = (e.payload->>'ordinal')::integer
          AND i.track_id = e.track_id
         WHERE e.session_id = $1
           AND e.type = 'track_served'
           AND e.payload->>'planVersion' = $2::text
           AND e.payload->>'planVersionId' = $3
         ORDER BY i.ordinal DESC
         LIMIT 1`,
        [sessionId, input.expectedPlanVersion, input.planVersionId],
      );
      const currentCursor = current.rows[0] as Pick<VibePlanItem, 'track_id' | 'ordinal'> | undefined;
      if (currentCursor?.track_id !== input.trackId || currentCursor.ordinal !== input.ordinal) {
        throw new Error('Vibe plan item is not the current served cursor for this session revision');
      }

      const payload = {
        kind: 'unplayable_plan_item',
        planVersion: input.expectedPlanVersion,
        planVersionId: input.planVersionId,
        ordinal: input.ordinal,
        trackId: input.trackId,
      };
      const playbackError = await client.query(
        `INSERT INTO vibe_events (client_event_id, session_id, user_id, track_id, type, occurred_at, payload)
         VALUES ($1::uuid, $2, $3, $4::uuid, 'playback_error', NOW(), $5::jsonb)
         RETURNING id`,
        [input.eventId, sessionId, userId, input.trackId, JSON.stringify(payload)],
      );
      const eventId = playbackError.rows[0]?.id as string | undefined;
      if (!eventId) throw new Error('Vibe playback-error event could not be recorded');

      const item = await client.query(
        `WITH next_item AS (
           SELECT i.plan_version_id, i.ordinal FROM vibe_plan_items i
           WHERE i.plan_version_id = $1 AND NOT i.committed ORDER BY i.ordinal ASC LIMIT 1 FOR UPDATE
         )
         UPDATE vibe_plan_items i SET committed = true
         FROM next_item n WHERE i.plan_version_id = n.plan_version_id AND i.ordinal = n.ordinal
         RETURNING i.*`,
        [plan.id],
      );
      const replacement = item.rows[0] as VibePlanItem | undefined;
      if (replacement) {
        await client.query(
          `INSERT INTO vibe_events (session_id, user_id, track_id, type, occurred_at, payload)
           VALUES ($1, $2, $3, 'track_served', NOW(), $4::jsonb)`,
          [sessionId, userId, replacement.track_id, JSON.stringify({
            planVersion: plan.version,
            planVersionId: replacement.plan_version_id,
            ordinal: replacement.ordinal,
          })],
        );
      }
      await client.query(
        `UPDATE vibe_events SET payload = $2::jsonb WHERE id = $1`,
        [eventId, JSON.stringify({
          ...payload,
          advancedTo: replacement
            ? { planVersionId: replacement.plan_version_id, ordinal: replacement.ordinal }
            : null,
        })],
      );
      await client.query(`UPDATE vibe_sessions SET last_event_at = NOW() WHERE id = $1`, [sessionId]);
      return { item: replacement ?? null, stale: false };
    });
  }

  /** Read a specific plan revision, or the latest revision for a session. */
  async getVibePlan(sessionId: string, userId: string, version?: number): Promise<VibePlan | null> {
    const res = await this.pgClient.query(
      `SELECT p.*, i.plan_version_id AS item_plan_version_id, i.ordinal, i.track_id,
              i.slot_role, i.candidate_source, i.score, i.score_breakdown,
              i.explanation, i.committed
       FROM vibe_plan_versions p
       JOIN vibe_sessions s ON s.id = p.session_id
       LEFT JOIN vibe_plan_items i ON i.plan_version_id = p.id
       WHERE p.session_id = $1 AND s.user_id = $2
         AND (
           ($3::integer IS NOT NULL AND p.version = $3)
           OR ($3::integer IS NULL AND p.version = (
             SELECT MAX(version) FROM vibe_plan_versions WHERE session_id = $1
           ))
         )
       ORDER BY i.ordinal ASC`,
      [sessionId, userId, version ?? null]
    );
    if (!res.rows[0]) return null;

    const first = res.rows[0] as VibePlanVersionRow;
    const plan: VibePlan = {
      id: first.id,
      session_id: first.session_id,
      version: first.version,
      reason: first.reason,
      state_snapshot: first.state_snapshot,
      objective_snapshot: first.objective_snapshot,
      created_at: first.created_at,
      items: [],
    };
    for (const row of res.rows as VibePlanVersionRow[]) {
      if (!row.item_plan_version_id) continue;
      plan.items.push({
        plan_version_id: row.item_plan_version_id,
        ordinal: row.ordinal!,
        track_id: row.track_id!,
        slot_role: row.slot_role,
        candidate_source: row.candidate_source!,
        score: row.score!,
        score_breakdown: row.score_breakdown!,
        explanation: row.explanation,
        committed: row.committed!,
      });
    }
    return plan;
  }

  /** Return the replacement revision caused by a material feedback event. */
  async getVibePlanForFeedbackEvent(sessionId: string, userId: string, eventId: string): Promise<VibePlan | null> {
    const res = await this.pgClient.query(
      `SELECT version
       FROM vibe_plan_versions p
       JOIN vibe_sessions s ON s.id = p.session_id
       WHERE p.session_id = $1
         AND s.user_id = $2
         AND p.objective_snapshot->>'feedbackEventId' = $3
       ORDER BY p.version DESC
       LIMIT 1`,
      [sessionId, userId, eventId],
    );
    const version = res.rows[0]?.version as number | undefined;
    return version === undefined ? null : this.getVibePlan(sessionId, userId, version);
  }

  /**
   * Tracks exposed by a durable session are never eligible for another
   * revision of that same session. This includes served items and every
   * explicit feedback target, not just completed play history.
   */
  async getVibeSessionTrackIds(sessionId: string, userId: string): Promise<string[]> {
    const res = await this.pgClient.query(
      `SELECT DISTINCT e.track_id
       FROM vibe_events e
       JOIN vibe_sessions s ON s.id = e.session_id
       WHERE e.session_id = $1
         AND s.user_id = $2
         AND e.track_id IS NOT NULL`,
      [sessionId, userId],
    );
    return res.rows.map((row: { track_id: string }) => row.track_id);
  }

  /**
   * Upsert a diversity budget for a user.
   */
  async upsertDiversityBudget(budget: {
    user_id: string;
    dimension: string;
    budget_share: number;
    horizon_min: number;
  }): Promise<void> {
    await this.pgClient.query(
      `INSERT INTO diversity_budgets (user_id, dimension, budget_share, horizon_min)
       VALUES ($1, $2, $3, $4)
       ON CONFLICT (user_id, dimension, horizon_min)
       DO UPDATE SET budget_share = $3`,
      [budget.user_id, budget.dimension, budget.budget_share, budget.horizon_min]
    );
  }

  /**
   * Seed default diversity budgets for a new user.
   */
  async seedDefaultDiversityBudgets(userId: string): Promise<void> {
    const defaults: { dimension: string; share: number; horizon: number }[] = [
      { dimension: 'artist', share: 0.20, horizon: 30 },
      { dimension: 'genre', share: 0.40, horizon: 30 },
      { dimension: 'language', share: 0.60, horizon: 30 },
      { dimension: 'instrumental', share: 0.10, horizon: 30 },
      { dimension: 'new_artist', share: 0.15, horizon: 60 },
      { dimension: 'favorite', share: 0.25, horizon: 60 },
    ];

    for (const d of defaults) {
      await this.upsertDiversityBudget({
        user_id: userId,
        dimension: d.dimension,
        budget_share: d.share,
        horizon_min: d.horizon,
      });
    }
  }

  /** Watermark for refreshClaimFusion() — last-seen MAX(claims.last_reinforced_at). */
  private lastClaimsWatermark: string | null = null;

  /**
   * Refresh the claim_fusion materialised view. Called on a periodic timer so
   * the graph's read path stays current with new claims. Skips the (fairly
   * expensive, shared-connection-stalling) REFRESH when claims haven't
   * changed since the last tick — a NOTIFY trigger exists on the claims
   * table but nothing LISTENs for it yet, so this cheap watermark check
   * stands in for that. last_reinforced_at is bumped by both inserts
   * (DEFAULT NOW()) and upsertClaim's ON CONFLICT UPDATE, so it tracks all
   * claim mutations that would change the view's output.
   * CONCURRENTLY requires the unique index (idx_claim_fusion_pk),
   * which the 20260708_materialize_claim_fusion migration creates.
   */
  async refreshClaimFusion(): Promise<void> {
    try {
      const watermarkRes = await this.pgClient.query<{ max: string | null }>(
        'SELECT MAX(last_reinforced_at) AS max FROM claims'
      );
      const watermark = watermarkRes.rows[0]?.max ?? null;
      if (watermark !== null && watermark === this.lastClaimsWatermark) {
        return; // no claim changes since the last refresh — skip it
      }
      await this.pgClient.query('SELECT refresh_claim_fusion()');
      this.lastClaimsWatermark = watermark;
    } catch (err) {
      // Non-fatal: the MV may not exist yet on first boot before
      // migrations run. Log and move on; the next tick will retry.
      console.error('[DB] refresh_claim_fusion failed:', err);
    }
  }

  /**
   * Decay all listener beliefs whose last_decayed_at is older than 1
   * hour. Implements the decay formula from spec §B.4:
   *   value *= 0.5 ^ (elapsed / halflife)
   *   confidence *= 0.5 ^ (elapsed / halflife)
   * Halflife is per-profile (longterm=365d, obsession=14d, discovery=30d,
   * negative=180d, contextual=7d). The 'forgotten' profile is excluded
   * — it is fully derived nightly by deriveForgottenProfile(), not
   * decayed.
   */
  async decayBeliefs(): Promise<number> {
    const res = await this.pgClient.query(`
      -- NOTE: this MUST be a VALUES list, not "SELECT profile, CASE profile ...",
      -- which has no FROM clause and is rejected by Postgres with 42703
      -- (column "profile" does not exist) on every single run.
      WITH halflives(profile, halflife_sec) AS (
        VALUES
          ('longterm',   365 * 86400),
          ('obsession',   14 * 86400),
          ('discovery',   30 * 86400),
          ('negative',   180 * 86400),
          ('contextual',   7 * 86400)
      )
      UPDATE listener_beliefs lb
      SET value = GREATEST(-1.0, LEAST(1.0, lb.value * POWER(0.5,
              EXTRACT(EPOCH FROM (NOW() - lb.last_decayed_at)) / h.halflife_sec))),
          confidence = GREATEST(0, LEAST(1.0, lb.confidence * POWER(0.5,
              EXTRACT(EPOCH FROM (NOW() - lb.last_decayed_at)) / h.halflife_sec))),
          last_decayed_at = NOW()
      FROM halflives h
      WHERE lb.profile = h.profile
        AND lb.profile <> 'forgotten'
        AND lb.last_decayed_at < NOW() - INTERVAL '1 hour'
    `);
    return res.rowCount ?? 0;
  }

  /**
   * Derive the 'forgotten' profile nightly (spec §B.2):
   *   longterm affinity > 0.3 AND not reinforced in 90+ days.
   * Wipes and repopulates — 'forgotten' is fully derived, not evidence-fed.
   */
  async deriveForgottenProfile(): Promise<number> {
    await this.pgClient.query(
      `DELETE FROM listener_beliefs WHERE profile = 'forgotten'`
    );
    const res = await this.pgClient.query(`
      INSERT INTO listener_beliefs
        (user_id, profile, entity_type, entity_id, dimension, value,
         confidence, evidence_count, last_reinforced_at, last_decayed_at)
      SELECT user_id, 'forgotten', entity_type, entity_id, dimension,
             value, confidence, evidence_count, last_reinforced_at, NOW()
      FROM listener_beliefs
      WHERE profile = 'longterm'
        AND dimension = 'affinity'
        AND value > 0.3
        AND last_reinforced_at < NOW() - INTERVAL '90 days'
      ON CONFLICT (user_id, profile, entity_type, entity_id, dimension)
      DO UPDATE SET
        value = EXCLUDED.value,
        confidence = EXCLUDED.confidence,
        evidence_count = EXCLUDED.evidence_count,
        last_reinforced_at = EXCLUDED.last_reinforced_at
    `);
    return res.rowCount ?? 0;
  }
}