WARP

# ── Data portability (Art. 20)

def export_user(repo, user_id) do
  user = repo.get!(User, user_id)
  acct_ids =
    repo.all(
      from a in Account,
      where: a.user_id == ^user_id,
      select: a.id
    )

  %{
    user: user,
    accounts: repo.all(
      from a in Account,
      where: a.user_id == ^user_id),
    transactions: repo.all(
      from t in Transaction,
      where: t.account_id in ^acct_ids),
    kyc: repo.all(
      from k in KycDocument,
      where: k.user_id == ^user_id),
    sessions: repo.all(
      from s in Session,
      where: s.user_id == ^user_id),
    audit: repo.all(
      from a in AuditLog,
      where: a.user_id == ^user_id),
  }
end

# ── Right to erasure (Art. 17)
# Order matters (FK constraints)

def delete_user(repo, user_id) do
  acct_ids =
    repo.all(
      from a in Account,
      where: a.user_id == ^user_id,
      select: a.id
    )

  Ecto.Multi.new()
  |> Ecto.Multi.delete_all(
    :intents,
    from(i in TransferIntent,
      where: i.from_acct in ^acct_ids
        or i.to_acct in ^acct_ids))
  |> Ecto.Multi.delete_all(
    :audit,
    from(a in AuditLog,
      where: a.user_id == ^user_id))
  |> Ecto.Multi.delete_all(
    :sessions,
    from(s in Session,
      where: s.user_id == ^user_id))
  |> Ecto.Multi.delete_all(
    :kyc,
    from(k in KycDocument,
      where: k.user_id == ^user_id))
  |> Ecto.Multi.delete_all(
    :txns,
    from(t in Transaction,
      where: t.account_id in ^acct_ids))
  |> Ecto.Multi.delete_all(
    :accounts,
    from(a in Account,
      where: a.user_id == ^user_id))
  |> Ecto.Multi.delete_all(
    :user,
    from(u in User,
      where: u.id == ^user_id))
  |> repo.transaction()
end

# ── Purge from replicas ───────

def purge_replicas(user_id) do
  Enum.each(replica_repos(), fn repo ->
    delete_user(repo, user_id)
  end)
end

# ── Advance replication slots ─

def advance_slots(repo) do
  {:ok, %{rows: slots}} =
    Ecto.Adapters.SQL.query!(repo,
      "SELECT slot_name
       FROM pg_replication_slots
       WHERE active = true")

  Enum.each(slots, fn [name] ->
    Ecto.Adapters.SQL.query!(repo,
      "SELECT
         pg_logical_slot_get_changes(
           $1, NULL, NULL)",
      [name])
  end)
end

# ── Force WAL rotation ────────

def rotate_wal(repo) do
  Ecto.Adapters.SQL.query!(
    repo, "SELECT pg_switch_wal()")
  Ecto.Adapters.SQL.query!(
    repo, "CHECKPOINT")
end

# ── Log for DPO audit ─────────

def log_deletion(repo, user_id) do
  repo.insert!(%GdprDeletionLog{
    user_id: user_id,
    tables_purged: 7,
    replicas_purged:
      length(replica_repos()),
    wal_rotated: true,
    slots_advanced: true
  })
end

# ── Run it all ────────────────

def gdpr_erase(user_id) do
  export = export_user(EURepo, user_id)
  {:ok, _} = delete_user(EURepo, user_id)
  purge_replicas(user_id)
  advance_slots(EURepo)
  rotate_wal(EURepo)
  log_deletion(EURepo, user_id)
  {:ok, export}
end

// ── Data portability (Art. 20)

async function exportUser(
  db: DrizzleDb, userId: string
) {
  const acctIds = (await db
    .select({ id: accounts.id })
    .from(accounts)
    .where(eq(accounts.userId, userId))
  ).map(a => a.id)

  return {
    user: await db.select().from(users)
      .where(eq(users.id, userId)),
    accounts: await db.select()
      .from(accounts)
      .where(eq(accounts.userId, userId)),
    transactions: await db.select()
      .from(transactions)
      .where(
        inArray(transactions.accountId,
          acctIds)),
    kyc: await db.select()
      .from(kycDocuments)
      .where(eq(kycDocuments.userId,
        userId)),
    sessions: await db.select()
      .from(sessions)
      .where(eq(sessions.userId, userId)),
    audit: await db.select()
      .from(auditLog)
      .where(eq(auditLog.userId, userId)),
  }
}

// ── Right to erasure (Art. 17)
// Order matters (FK constraints)

async function deleteUser(
  db: DrizzleDb, userId: string
) {
  const acctIds = (await db
    .select({ id: accounts.id })
    .from(accounts)
    .where(eq(accounts.userId, userId))
  ).map(a => a.id)

  await db.transaction(async (tx) => {
    await tx.delete(transferIntents)
      .where(or(
        inArray(transferIntents.fromAcct,
          acctIds),
        inArray(transferIntents.toAcct,
          acctIds)))
    await tx.delete(auditLog)
      .where(eq(auditLog.userId, userId))
    await tx.delete(sessions)
      .where(eq(sessions.userId, userId))
    await tx.delete(kycDocuments)
      .where(eq(kycDocuments.userId,
        userId))
    await tx.delete(transactions)
      .where(
        inArray(transactions.accountId,
          acctIds))
    await tx.delete(accounts)
      .where(eq(accounts.userId, userId))
    await tx.delete(users)
      .where(eq(users.id, userId))
  })
}

// ── Purge from replicas ───────

async function purgeReplicas(
  userId: string
) {
  for (const replica of replicaDbs) {
    await deleteUser(replica, userId)
  }
}

// ── Advance replication slots ─

async function advanceSlots(
  db: DrizzleDb
) {
  const { rows: slots } = await db
    .execute(sql`
      SELECT slot_name
      FROM pg_replication_slots
      WHERE active = true`)

  for (const s of slots) {
    await db.execute(sql`
      SELECT pg_logical_slot_get_changes(
        ${s.slot_name}, NULL, NULL
      )`)
  }
}

// ── Force WAL rotation ────────

async function rotateWal(db: DrizzleDb) {
  await db.execute(
    sql`SELECT pg_switch_wal()`)
  await db.execute(sql`CHECKPOINT`)
}

// ── Run it all ───────────────

async function gdprErase(userId: string) {
  const exported = await exportUser(
    euDb, userId)
  await deleteUser(euDb, userId)
  await purgeReplicas(userId)
  await advanceSlots(euDb)
  await rotateWal(euDb)
  await euDb.insert(gdprDeletionLog)
    .values({
      userId,
      tablesPurged: 7,
      replicasPurged: replicaDbs.length,
      walRotated: true,
      slotsAdvanced: true,
    })
  return exported
}