-- Desktop ↔ mobile sync parity.
--
-- The desktop app keeps a local SQLite copy and mirrors it through
-- CloudSyncService. Mobile and web read and write Supabase directly. Before this
-- migration the mirror could not see deletions made on another device, could
-- not tell whether a desktop write changed a history row that mobile holds
-- with an expected revision, and could not pull edited meeting memos at all.
--
-- 1. sync_tombstones: every owner-scoped delete on a synced table leaves a
--    tombstone the owner can read, so an offline device applies it later.
-- 2. history.revision moves on any content change, including writes that do
--    not send a revision (desktop upserts), so mobile's optimistic
--    concurrency sees desktop edits.
-- 3. meeting_memos.updated_at so memo edits are pullable by cursor.
-- 4. (user_id, updated_at, id) cursor indexes for keyset pagination.
-- 5. Desktop-facing template upsert/delete RPCs keyed by the client id, built
--    on the existing revision-checked RPCs.
-- 6. Realtime for tombstones and devices (remote sign-out of a revoked
--    desktop).

BEGIN;

-- 1) Tombstones ----------------------------------------------------------------
-- No foreign key to auth.users on purpose: an account deletion cascades into
-- the synced tables and fires the tombstone trigger for a user row that is
-- already gone in the same statement. A foreign key would abort the account
-- deletion. Orphaned tombstones only carry ids and are pruned below.
CREATE TABLE IF NOT EXISTS public.sync_tombstones (
  id          bigserial PRIMARY KEY,
  user_id     uuid NOT NULL,
  table_name  text NOT NULL CHECK (table_name IN (
    'history',
    'dictionary',
    'meetings',
    'meeting_memos',
    'meeting_documents',
    'memo_tags',
    'custom_instructions',
    'user_templates'
  )),
  row_id      uuid NOT NULL,
  deleted_at  timestamptz NOT NULL DEFAULT clock_timestamp()
);

CREATE INDEX IF NOT EXISTS idx_sync_tombstones_user_deleted
  ON public.sync_tombstones(user_id, deleted_at, id);

ALTER TABLE public.sync_tombstones ENABLE ROW LEVEL SECURITY;

DROP POLICY IF EXISTS sync_tombstones_read_own ON public.sync_tombstones;
CREATE POLICY sync_tombstones_read_own
  ON public.sync_tombstones FOR SELECT TO authenticated
  USING (user_id = auth.uid());

REVOKE ALL ON TABLE public.sync_tombstones FROM PUBLIC, anon, authenticated;
GRANT SELECT ON TABLE public.sync_tombstones TO authenticated;
REVOKE ALL ON SEQUENCE public.sync_tombstones_id_seq FROM PUBLIC, anon, authenticated;

CREATE OR REPLACE FUNCTION public.record_sync_tombstone_v1()
RETURNS trigger
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, public
AS $$
BEGIN
  INSERT INTO public.sync_tombstones(user_id, table_name, row_id)
  VALUES (OLD.user_id, TG_TABLE_NAME, OLD.id);
  RETURN OLD;
END;
$$;

REVOKE ALL ON FUNCTION public.record_sync_tombstone_v1() FROM PUBLIC, anon, authenticated;

DO $$
DECLARE
  t text;
BEGIN
  FOREACH t IN ARRAY ARRAY[
    'history',
    'dictionary',
    'meetings',
    'meeting_memos',
    'meeting_documents',
    'memo_tags',
    'custom_instructions',
    'user_templates'
  ]
  LOOP
    EXECUTE format('DROP TRIGGER IF EXISTS record_sync_tombstone_v1 ON public.%I', t);
    EXECUTE format(
      'CREATE TRIGGER record_sync_tombstone_v1 AFTER DELETE ON public.%I '
      'FOR EACH ROW EXECUTE FUNCTION public.record_sync_tombstone_v1()',
      t
    );
  END LOOP;
END $$;

-- Tombstones older than the retention window are useless to a client whose
-- cursor is older still: such a client must do a full resync anyway. Rows of
-- deleted accounts are removed at the same time.
CREATE OR REPLACE FUNCTION public.prune_sync_tombstones_v1(p_retention interval DEFAULT interval '180 days')
RETURNS bigint
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, public, auth
AS $$
DECLARE
  removed bigint;
BEGIN
  DELETE FROM public.sync_tombstones AS tombstone
  WHERE tombstone.deleted_at < clock_timestamp() - p_retention
     OR NOT EXISTS (SELECT 1 FROM auth.users AS account WHERE account.id = tombstone.user_id);
  GET DIAGNOSTICS removed = ROW_COUNT;
  RETURN removed;
END;
$$;

REVOKE ALL ON FUNCTION public.prune_sync_tombstones_v1(interval) FROM PUBLIC, anon, authenticated;
GRANT EXECUTE ON FUNCTION public.prune_sync_tombstones_v1(interval) TO service_role;

-- 2) history.revision follows content changes -----------------------------------
-- Mobile updates with `revision = expected + 1` and a revision predicate, so a
-- write that already moved the revision is left alone. A write that did not
-- (the desktop mirror upserts full rows) bumps it only when user-visible
-- content changed, so a re-push of identical content never forces a false
-- conflict on another device.
CREATE OR REPLACE FUNCTION public.bump_history_revision_v1()
RETURNS trigger
LANGUAGE plpgsql
SET search_path = ''
AS $$
BEGIN
  IF NEW.revision IS NOT DISTINCT FROM OLD.revision
     AND (
       NEW.title,
       NEW.original_text,
       NEW.polished_text,
       NEW.summary_text,
       NEW.is_favorite,
       NEW.status,
       NEW.mode
     ) IS DISTINCT FROM (
       OLD.title,
       OLD.original_text,
       OLD.polished_text,
       OLD.summary_text,
       OLD.is_favorite,
       OLD.status,
       OLD.mode
     ) THEN
    NEW.revision := OLD.revision + 1;
  END IF;
  RETURN NEW;
END;
$$;

DROP TRIGGER IF EXISTS bump_history_revision_v1 ON public.history;
CREATE TRIGGER bump_history_revision_v1
  BEFORE UPDATE ON public.history
  FOR EACH ROW EXECUTE FUNCTION public.bump_history_revision_v1();

-- 3) meeting_memos.updated_at ----------------------------------------------------
ALTER TABLE public.meeting_memos ADD COLUMN IF NOT EXISTS updated_at timestamptz;
UPDATE public.meeting_memos SET updated_at = created_at WHERE updated_at IS NULL;
ALTER TABLE public.meeting_memos
  ALTER COLUMN updated_at SET DEFAULT now(),
  ALTER COLUMN updated_at SET NOT NULL;

DROP TRIGGER IF EXISTS set_updated_at_meeting_memos ON public.meeting_memos;
CREATE TRIGGER set_updated_at_meeting_memos
  BEFORE UPDATE ON public.meeting_memos
  FOR EACH ROW EXECUTE FUNCTION public.moddatetime();

-- 4) Cursor indexes --------------------------------------------------------------
CREATE INDEX IF NOT EXISTS idx_history_sync_cursor
  ON public.history(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_dictionary_sync_cursor
  ON public.dictionary(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_meetings_sync_cursor
  ON public.meetings(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_meeting_memos_sync_cursor
  ON public.meeting_memos(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_meeting_documents_sync_cursor
  ON public.meeting_documents(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_custom_instructions_sync_cursor
  ON public.custom_instructions(user_id, updated_at, id);
CREATE INDEX IF NOT EXISTS idx_user_templates_sync_cursor
  ON public.user_templates(user_id, updated_at, id);

-- 5) Client-keyed template sync ---------------------------------------------------
-- The desktop creates templates offline under its own uuid. These wrappers keep
-- that id and reuse the revision-checked mobile RPCs for every update/delete
-- rule (builtin immutability, selection fallback), so there is one mutation
-- contract per template.
CREATE OR REPLACE FUNCTION public.sync_upsert_user_template_v1(
  p_id uuid,
  p_template_kind text,
  p_name text,
  p_description text DEFAULT NULL,
  p_fields jsonb DEFAULT '[]'::jsonb,
  p_output_format text DEFAULT NULL,
  p_system_prompt text DEFAULT NULL
)
RETURNS public.user_templates
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, public, auth
AS $$
DECLARE
  actor_id uuid := auth.uid();
  current_template public.user_templates;
  saved public.user_templates;
BEGIN
  IF actor_id IS NULL THEN
    RAISE EXCEPTION 'authentication_required' USING ERRCODE = '42501';
  END IF;
  IF p_id IS NULL THEN
    RAISE EXCEPTION 'template_id_required' USING ERRCODE = '22023';
  END IF;
  IF p_template_kind NOT IN ('dictation', 'meeting_document') THEN
    RAISE EXCEPTION 'invalid_template_kind' USING ERRCODE = '22023';
  END IF;

  SELECT * INTO current_template
  FROM public.user_templates
  WHERE id = p_id
  FOR UPDATE;

  IF current_template.id IS NOT NULL THEN
    IF current_template.user_id <> actor_id THEN
      RAISE EXCEPTION 'template_not_found' USING ERRCODE = 'P0002';
    END IF;
    IF current_template.template_kind <> p_template_kind THEN
      RAISE EXCEPTION 'template_kind_immutable' USING ERRCODE = '22023';
    END IF;
    saved := public.update_user_template_v1(
      p_id,
      current_template.revision,
      p_name,
      p_description,
      p_fields,
      p_output_format,
      p_system_prompt
    );
    RETURN saved;
  END IF;

  INSERT INTO public.user_templates(
    id, user_id, template_kind, name, description, fields, output_format,
    template_type, system_prompt, is_builtin
  )
  VALUES (
    p_id,
    actor_id,
    p_template_kind,
    trim(p_name),
    nullif(trim(p_description), ''),
    CASE WHEN p_template_kind = 'dictation' THEN p_fields ELSE '[]'::jsonb END,
    CASE WHEN p_template_kind = 'dictation' THEN p_output_format ELSE NULL END,
    CASE WHEN p_template_kind = 'meeting_document' THEN 'custom' ELSE NULL END,
    CASE WHEN p_template_kind = 'meeting_document' THEN p_system_prompt ELSE NULL END,
    false
  )
  RETURNING * INTO saved;

  RETURN saved;
END;
$$;

REVOKE ALL ON FUNCTION public.sync_upsert_user_template_v1(uuid, text, text, text, jsonb, text, text)
  FROM PUBLIC, anon;
GRANT EXECUTE ON FUNCTION public.sync_upsert_user_template_v1(uuid, text, text, text, jsonb, text, text)
  TO authenticated;

-- Deleting a template another device already deleted is success.
CREATE OR REPLACE FUNCTION public.sync_delete_user_template_v1(p_id uuid)
RETURNS boolean
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, public, auth
AS $$
DECLARE
  actor_id uuid := auth.uid();
  current_revision bigint;
BEGIN
  IF actor_id IS NULL THEN
    RAISE EXCEPTION 'authentication_required' USING ERRCODE = '42501';
  END IF;

  SELECT revision INTO current_revision
  FROM public.user_templates
  WHERE id = p_id AND user_id = actor_id
  FOR UPDATE;

  IF current_revision IS NULL THEN
    RETURN false;
  END IF;

  RETURN public.delete_user_template_v1(p_id, current_revision);
END;
$$;

REVOKE ALL ON FUNCTION public.sync_delete_user_template_v1(uuid) FROM PUBLIC, anon;
GRANT EXECUTE ON FUNCTION public.sync_delete_user_template_v1(uuid) TO authenticated;

-- 6) Realtime -----------------------------------------------------------------------
DO $$
DECLARE
  t text;
BEGIN
  FOREACH t IN ARRAY ARRAY['sync_tombstones', 'devices', 'memo_tags', 'custom_instructions', 'user_templates']
  LOOP
    IF NOT EXISTS (
      SELECT 1
      FROM pg_publication_tables
      WHERE pubname = 'supabase_realtime'
        AND schemaname = 'public'
        AND tablename = t
    ) THEN
      EXECUTE format('ALTER PUBLICATION supabase_realtime ADD TABLE public.%I', t);
    END IF;
  END LOOP;
END $$;

-- 7) Portable archive keeps its v1 shape ---------------------------------------
-- export_account_portability serialised whole rows with to_jsonb, while the
-- v1 archive contract (server restore and the mobile parser) accepts an exact
-- key set. Columns added after the archive was frozen (meetings creation
-- metadata, meeting_documents template/idempotency, meeting_memos.updated_at
-- above) made every archive with such rows fail validation. Project each
-- dataset back onto its v1 keys.
CREATE OR REPLACE FUNCTION public.export_account_portability()
RETURNS TABLE (
  canonical_payload text,
  checksum text,
  exported_at timestamptz,
  row_count integer
)
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = ''
AS $$
DECLARE
  current_user_id uuid := auth.uid();
  payload jsonb;
  payload_text text;
  dataset_rows integer;
  exported_timestamp timestamptz := clock_timestamp();
BEGIN
  IF current_user_id IS NULL THEN
    RAISE EXCEPTION 'authentication_required' USING ERRCODE = '42501';
  END IF;

  SELECT
      (SELECT count(*) FROM public.dictionary WHERE user_id = current_user_id)
    + (SELECT count(*) FROM public.history WHERE user_id = current_user_id)
    + (SELECT count(*) FROM public.meetings WHERE user_id = current_user_id AND team_id IS NULL)
    + (
        SELECT count(*)
          FROM public.transcripts AS transcript
          JOIN public.meetings AS meeting ON meeting.id = transcript.meeting_id
         WHERE meeting.user_id = current_user_id
           AND meeting.team_id IS NULL
      )
    + (
        SELECT count(*)
          FROM public.meeting_memos AS memo
          JOIN public.meetings AS meeting ON meeting.id = memo.meeting_id
         WHERE memo.user_id = current_user_id
           AND meeting.user_id = current_user_id
           AND meeting.team_id IS NULL
      )
    + (
        SELECT count(*)
          FROM public.meeting_documents AS document
          JOIN public.meetings AS meeting ON meeting.id = document.meeting_id
         WHERE document.user_id = current_user_id
           AND meeting.user_id = current_user_id
           AND meeting.team_id IS NULL
      )
    + (
        SELECT count(*)
          FROM public.custom_instructions
         WHERE user_id = current_user_id
           AND builtin_key IS NULL
      )
    INTO dataset_rows;

  IF dataset_rows > 10000 THEN
    RAISE EXCEPTION 'export_row_limit_exceeded' USING ERRCODE = '54000';
  END IF;

  payload := jsonb_build_object(
    'format', 'd3ro-account-portability',
    'schema_version', 1,
    'exported_at', exported_timestamp,
    'owner_id', current_user_id,
    'source', 'cloud',
    'account', jsonb_build_object(
      'profile', coalesce((
        SELECT jsonb_build_object(
          'id', profile.id,
          'name', profile.name,
          'avatar_url', profile.avatar_url,
          'locale', profile.locale,
          'tier', profile.tier,
          'created_at', profile.created_at,
          'updated_at', profile.updated_at
        )
        FROM public.profiles AS profile
        WHERE profile.id = current_user_id
      ), 'null'::jsonb),
      'settings', coalesce((
        SELECT jsonb_build_object(
          'theme_mode', settings.theme_mode,
          'locale', settings.locale,
          'haptic_enabled', settings.haptic_enabled,
          'auto_polish_enabled', settings.auto_polish_enabled,
          'preferred_stt_model', settings.preferred_stt_model,
          'preferred_llm_model', settings.preferred_llm_model,
          'onboarding_version', settings.onboarding_version,
          'tutorial_completed_at', settings.tutorial_completed_at,
          'revision', settings.revision,
          'updated_at', settings.updated_at
        )
        FROM public.user_settings AS settings
        WHERE settings.user_id = current_user_id
      ), 'null'::jsonb),
      'subscription', coalesce((
        SELECT jsonb_build_object(
          'tier', subscription.tier,
          'provider', subscription.provider,
          'status', subscription.status,
          'current_period_start', subscription.current_period_start,
          'current_period_end', subscription.current_period_end,
          'cancel_at', subscription.cancel_at,
          'auto_renewing', subscription.auto_renewing,
          'updated_at', subscription.updated_at
        )
        FROM public.subscriptions AS subscription
        WHERE subscription.user_id = current_user_id
      ), 'null'::jsonb)
    ),
    'datasets', jsonb_build_object(
      'dictionary', coalesce((
        SELECT jsonb_agg(to_jsonb(dictionary) ORDER BY dictionary.created_at, dictionary.id)
        FROM public.dictionary AS dictionary
        WHERE dictionary.user_id = current_user_id
      ), '[]'::jsonb),
      'history', coalesce((
        SELECT jsonb_agg(to_jsonb(history) - 'audio_storage_key' ORDER BY history.created_at, history.id)
        FROM public.history AS history
        WHERE history.user_id = current_user_id
      ), '[]'::jsonb),
      'meetings', coalesce((
        SELECT jsonb_agg(to_jsonb(meeting) - ARRAY['audio_storage_key', 'language', 'attendees', 'template_id', 'creation_idempotency_key', 'creation_request_hash']::text[] ORDER BY meeting.created_at, meeting.id)
        FROM public.meetings AS meeting
        WHERE meeting.user_id = current_user_id
          AND meeting.team_id IS NULL
      ), '[]'::jsonb),
      'transcripts', coalesce((
        SELECT jsonb_agg(to_jsonb(transcript) ORDER BY transcript.meeting_id, transcript.segment_index, transcript.id)
        FROM public.transcripts AS transcript
        JOIN public.meetings AS meeting ON meeting.id = transcript.meeting_id
        WHERE meeting.user_id = current_user_id
          AND meeting.team_id IS NULL
      ), '[]'::jsonb),
      'meeting_memos', coalesce((
        SELECT jsonb_agg(to_jsonb(memo) - 'updated_at' ORDER BY memo.meeting_id, memo.timestamp_ms, memo.id)
        FROM public.meeting_memos AS memo
        JOIN public.meetings AS meeting ON meeting.id = memo.meeting_id
        WHERE memo.user_id = current_user_id
          AND meeting.user_id = current_user_id
          AND meeting.team_id IS NULL
      ), '[]'::jsonb),
      'meeting_documents', coalesce((
        SELECT jsonb_agg(to_jsonb(document) - ARRAY['template_id', 'generation_idempotency_key']::text[] ORDER BY document.meeting_id, document.created_at, document.id)
        FROM public.meeting_documents AS document
        JOIN public.meetings AS meeting ON meeting.id = document.meeting_id
        WHERE document.user_id = current_user_id
          AND meeting.user_id = current_user_id
          AND meeting.team_id IS NULL
      ), '[]'::jsonb),
      'custom_instructions', coalesce((
        SELECT jsonb_agg(to_jsonb(instruction) ORDER BY instruction.sort_order, instruction.created_at, instruction.id)
        FROM public.custom_instructions AS instruction
        WHERE instruction.user_id = current_user_id
          AND instruction.builtin_key IS NULL
      ), '[]'::jsonb)
    ),
    'exclusions', jsonb_build_array(
      'raw_audio',
      'storage_objects',
      'payment_credentials',
      'push_tokens'
    )
  );

  payload_text := payload::text;
  IF octet_length(payload_text) > 5242880 THEN
    RAISE EXCEPTION 'export_payload_too_large' USING ERRCODE = '54000';
  END IF;

  RETURN QUERY SELECT
    payload_text,
    encode(extensions.digest(convert_to(payload_text, 'UTF8'), 'sha256'), 'hex'),
    exported_timestamp,
    dataset_rows;
END;
$$;

COMMIT;