From 50fcd0bb495a6e18bf54a9a7d4a3d2c657cb2222 Mon Sep 17 00:00:00 2001 From: Timothy Quilling Date: Thu, 20 Nov 2025 22:18:28 -0500 Subject: [PATCH] feat: more hypertables --- consumer/src/db/actor.rs | 319 ++++++++++-------- consumer/src/db/allowlist.rs | 36 +- consumer/src/db/bulk_resolve/mod.rs | 20 +- consumer/src/db/operations/feed/helpers.rs | 1 - consumer/src/db/operations/graph.rs | 30 +- consumer/src/db/workers.rs | 33 +- consumer/src/workers/backfill/downloader.rs | 28 +- .../down.sql | 126 +++++++ .../up.sql | 274 +++++++++++++++ parakeet-db/src/schema.rs | 19 -- 10 files changed, 670 insertions(+), 216 deletions(-) create mode 100644 migrations/2025-11-20-223826_hypertable_compression/down.sql create mode 100644 migrations/2025-11-20-223826_hypertable_compression/up.sql diff --git a/consumer/src/db/actor.rs b/consumer/src/db/actor.rs index 62c1c3f5..bfd1f45c 100644 --- a/consumer/src/db/actor.rs +++ b/consumer/src/db/actor.rs @@ -14,6 +14,11 @@ pub async fn actor_upsert( account_created_at: Option<&DateTime>, time: DateTime, ) -> Result { + // Acquire advisory lock on DID to prevent races + let (table_id, key_id) = crate::database_writer::locking::table_record_lock("actors", did); + conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) + .await?; + // Allow allowlist states (synced, dirty, processing) to flow freely // Allow upgrading from partial to allowlist states // Never downgrade from allowlist states to partial @@ -23,136 +28,224 @@ pub async fn actor_upsert( match (status, handle, account_created_at) { (Some(status), Some(handle), Some(created_at)) => { - // All three provided - conn.execute( - "INSERT INTO actors (did, status, handle, sync_state, account_created_at, last_indexed) VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (did) DO UPDATE SET - status=EXCLUDED.status, - handle=EXCLUDED.handle, + // All three provided - try update first, then insert if needed + let updated = conn.execute( + "UPDATE actors SET + status=$2, + handle=$3, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $4 = 'partial'::actor_sync_state + THEN sync_state + ELSE $4 END, - account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at), - last_indexed=EXCLUDED.last_indexed", + account_created_at=COALESCE(account_created_at, $5), + last_indexed=$6 + WHERE did=$1", &[&did, &status, &handle, &sync_state, &created_at, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, status, handle, sync_state, account_created_at, last_indexed) + VALUES ($1, $2, $3, $4, $5, $6)", + &[&did, &status, &handle, &sync_state, &created_at, &time], + ) + .await + } else { + Ok(updated) + } } (Some(status), Some(handle), None) => { // Status and handle, no created_at - conn.execute( - "INSERT INTO actors (did, status, handle, sync_state, last_indexed) VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (did) DO UPDATE SET - status=EXCLUDED.status, - handle=EXCLUDED.handle, + let updated = conn.execute( + "UPDATE actors SET + status=$2, + handle=$3, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $4 = 'partial'::actor_sync_state + THEN sync_state + ELSE $4 END, - last_indexed=EXCLUDED.last_indexed", + last_indexed=$5 + WHERE did=$1", &[&did, &status, &handle, &sync_state, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, status, handle, sync_state, last_indexed) + VALUES ($1, $2, $3, $4, $5)", + &[&did, &status, &handle, &sync_state, &time], + ) + .await + } else { + Ok(updated) + } } (Some(status), None, Some(created_at)) => { // Status and created_at, no handle - conn.execute( - "INSERT INTO actors (did, status, sync_state, account_created_at, last_indexed) VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (did) DO UPDATE SET - status=EXCLUDED.status, + let updated = conn.execute( + "UPDATE actors SET + status=$2, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $3 = 'partial'::actor_sync_state + THEN sync_state + ELSE $3 END, - account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at), - last_indexed=EXCLUDED.last_indexed", + account_created_at=COALESCE(account_created_at, $4), + last_indexed=$5 + WHERE did=$1", &[&did, &status, &sync_state, &created_at, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, status, sync_state, account_created_at, last_indexed) + VALUES ($1, $2, $3, $4, $5)", + &[&did, &status, &sync_state, &created_at, &time], + ) + .await + } else { + Ok(updated) + } } (Some(status), None, None) => { // Only status provided - conn.execute( - "INSERT INTO actors (did, status, sync_state, last_indexed) VALUES ($1, $2, $3, $4) - ON CONFLICT (did) DO UPDATE SET - status=EXCLUDED.status, + let updated = conn.execute( + "UPDATE actors SET + status=$2, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $3 = 'partial'::actor_sync_state + THEN sync_state + ELSE $3 END, - last_indexed=EXCLUDED.last_indexed", + last_indexed=$4 + WHERE did=$1", &[&did, &status, &sync_state, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, status, sync_state, last_indexed) + VALUES ($1, $2, $3, $4)", + &[&did, &status, &sync_state, &time], + ) + .await + } else { + Ok(updated) + } } (None, Some(handle), Some(created_at)) => { // Handle and created_at, no status - conn.execute( - "INSERT INTO actors (did, handle, sync_state, account_created_at, last_indexed) VALUES ($1, $2, $3, $4, $5) - ON CONFLICT (did) DO UPDATE SET - handle=EXCLUDED.handle, + let updated = conn.execute( + "UPDATE actors SET + handle=$2, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $3 = 'partial'::actor_sync_state + THEN sync_state + ELSE $3 END, - account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at), - last_indexed=EXCLUDED.last_indexed", + account_created_at=COALESCE(account_created_at, $4), + last_indexed=$5 + WHERE did=$1", &[&did, &handle, &sync_state, &created_at, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, handle, sync_state, account_created_at, last_indexed) + VALUES ($1, $2, $3, $4, $5)", + &[&did, &handle, &sync_state, &created_at, &time], + ) + .await + } else { + Ok(updated) + } } (None, Some(handle), None) => { // Only handle provided - conn.execute( - "INSERT INTO actors (did, handle, sync_state, last_indexed) VALUES ($1, $2, $3, $4) - ON CONFLICT (did) DO UPDATE SET - handle=EXCLUDED.handle, + let updated = conn.execute( + "UPDATE actors SET + handle=$2, sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $3 = 'partial'::actor_sync_state + THEN sync_state + ELSE $3 END, - last_indexed=EXCLUDED.last_indexed", + last_indexed=$4 + WHERE did=$1", &[&did, &handle, &sync_state, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, handle, sync_state, last_indexed) + VALUES ($1, $2, $3, $4)", + &[&did, &handle, &sync_state, &time], + ) + .await + } else { + Ok(updated) + } } (None, None, Some(created_at)) => { // Only created_at provided - conn.execute( - "INSERT INTO actors (did, sync_state, account_created_at, last_indexed) VALUES ($1, $2, $3, $4) - ON CONFLICT (did) DO UPDATE SET + let updated = conn.execute( + "UPDATE actors SET sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $2 = 'partial'::actor_sync_state + THEN sync_state + ELSE $2 END, - account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at), - last_indexed=EXCLUDED.last_indexed", + account_created_at=COALESCE(account_created_at, $3), + last_indexed=$4 + WHERE did=$1", &[&did, &sync_state, &created_at, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, sync_state, account_created_at, last_indexed) + VALUES ($1, $2, $3, $4)", + &[&did, &sync_state, &created_at, &time], + ) + .await + } else { + Ok(updated) + } } (None, None, None) => { // Neither provided - just ensure actor exists with sync_state - conn.execute( - "INSERT INTO actors (did, sync_state, last_indexed) VALUES ($1, $2, $3) - ON CONFLICT (did) DO UPDATE SET + let updated = conn.execute( + "UPDATE actors SET sync_state=CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state + WHEN sync_state IN ('synced', 'dirty', 'processing') AND $2 = 'partial'::actor_sync_state + THEN sync_state + ELSE $2 END, - last_indexed=EXCLUDED.last_indexed", + last_indexed=$3 + WHERE did=$1", &[&did, &sync_state, &time], ) - .await + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, sync_state, last_indexed) + VALUES ($1, $2, $3)", + &[&did, &sync_state, &time], + ) + .await + } else { + Ok(updated) + } } } .wrap_err_with(|| format!("Failed to upsert actor {}", did)) @@ -250,15 +343,7 @@ pub async fn ensure_actor_id( INSERT INTO actors (did, status, handle, sync_state, last_indexed) SELECT $1, $2, $3, $4, $5 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - status = COALESCE(EXCLUDED.status, actors.status), - handle = COALESCE(EXCLUDED.handle, actors.handle), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id ) SELECT COALESCE( @@ -278,14 +363,7 @@ pub async fn ensure_actor_id( INSERT INTO actors (did, status, sync_state, last_indexed) SELECT $1, $2, $3, $4 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - status = COALESCE(EXCLUDED.status, actors.status), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id ) SELECT COALESCE( @@ -305,14 +383,7 @@ pub async fn ensure_actor_id( INSERT INTO actors (did, handle, sync_state, last_indexed) SELECT $1, $2, $3, $4 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - handle = COALESCE(EXCLUDED.handle, actors.handle), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id ) SELECT COALESCE( @@ -332,13 +403,7 @@ pub async fn ensure_actor_id( INSERT INTO actors (did, sync_state, last_indexed) SELECT $1, $2, $3 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id ) SELECT COALESCE( @@ -398,15 +463,7 @@ pub async fn ensure_actor_id_with_cache( INSERT INTO actors (did, status, handle, sync_state, last_indexed) SELECT $1, $2, $3, $4, $5 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - status = COALESCE(EXCLUDED.status, actors.status), - handle = COALESCE(EXCLUDED.handle, actors.handle), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id, sync_state ) SELECT @@ -426,14 +483,7 @@ pub async fn ensure_actor_id_with_cache( INSERT INTO actors (did, status, sync_state, last_indexed) SELECT $1, $2, $3, $4 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - status = COALESCE(EXCLUDED.status, actors.status), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id, sync_state ) SELECT @@ -453,14 +503,7 @@ pub async fn ensure_actor_id_with_cache( INSERT INTO actors (did, handle, sync_state, last_indexed) SELECT $1, $2, $3, $4 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - handle = COALESCE(EXCLUDED.handle, actors.handle), - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id, sync_state ) SELECT @@ -480,13 +523,7 @@ pub async fn ensure_actor_id_with_cache( INSERT INTO actors (did, sync_state, last_indexed) SELECT $1, $2, $3 WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET - sync_state = CASE - WHEN actors.sync_state IN ('synced', 'dirty', 'processing') AND EXCLUDED.sync_state = 'partial' - THEN actors.sync_state - ELSE EXCLUDED.sync_state - END, - last_indexed = EXCLUDED.last_indexed + ON CONFLICT DO NOTHING -- Never conflicts due to WHERE NOT EXISTS RETURNING id, sync_state ) SELECT diff --git a/consumer/src/db/allowlist.rs b/consumer/src/db/allowlist.rs index 0a3f4273..d6f83eb7 100644 --- a/consumer/src/db/allowlist.rs +++ b/consumer/src/db/allowlist.rs @@ -146,16 +146,32 @@ pub async fn add( ) -> Result { // Note: description parameter kept for API compatibility but is no longer stored // Admin notes should be kept in external documentation - let rows_affected = client - .execute( - "INSERT INTO actors (did, status, sync_state) - VALUES ($1, 'active'::actor_status, 'dirty'::actor_sync_state) - ON CONFLICT (did) DO UPDATE - SET sync_state = 'dirty'::actor_sync_state - WHERE actors.sync_state = 'partial'::actor_sync_state", - &[&did], - ) - .await?; + + // Check if actor already exists (no unique constraint on did anymore) + let exists = client + .query_opt("SELECT 1 FROM actors WHERE did=$1", &[&did]) + .await? + .is_some(); + + let rows_affected = if exists { + // Actor exists - try UPDATE (only if partial) + client + .execute( + "UPDATE actors SET sync_state = 'dirty'::actor_sync_state + WHERE did=$1 AND sync_state = 'partial'::actor_sync_state", + &[&did], + ) + .await? + } else { + // Actor doesn't exist - INSERT + client + .execute( + "INSERT INTO actors (did, status, sync_state) + VALUES ($1, 'active'::actor_status, 'dirty'::actor_sync_state)", + &[&did], + ) + .await? + }; Ok(rows_affected) } diff --git a/consumer/src/db/bulk_resolve/mod.rs b/consumer/src/db/bulk_resolve/mod.rs index 7bc83454..2f435fe4 100644 --- a/consumer/src/db/bulk_resolve/mod.rs +++ b/consumer/src/db/bulk_resolve/mod.rs @@ -152,10 +152,22 @@ pub async fn create_actor_stubs_bulk( let rows = conn .query( - "INSERT INTO actors (did, status, sync_state, last_indexed) - SELECT UNNEST($1::text[]), 'active', 'partial', NOW() - ON CONFLICT (did) DO NOTHING - RETURNING did, id", + "WITH input_dids AS ( + SELECT UNNEST($1::text[]) as did + ), + existing AS ( + SELECT a.did + FROM actors a + INNER JOIN input_dids i ON a.did = i.did + ), + inserted AS ( + INSERT INTO actors (did, status, sync_state, last_indexed) + SELECT i.did, 'active', 'partial', NOW() + FROM input_dids i + WHERE NOT EXISTS (SELECT 1 FROM existing e WHERE e.did = i.did) + RETURNING did, id + ) + SELECT did, id FROM inserted", &[&dids], ) .await?; diff --git a/consumer/src/db/operations/feed/helpers.rs b/consumer/src/db/operations/feed/helpers.rs index 6f6ce2d8..5c967533 100644 --- a/consumer/src/db/operations/feed/helpers.rs +++ b/consumer/src/db/operations/feed/helpers.rs @@ -73,7 +73,6 @@ pub async fn get_actor_id(conn: &C, did: &str) -> Result<(i32, INSERT INTO actors (did, status, sync_state, last_indexed) SELECT $1, 'active', 'partial', NOW() WHERE NOT EXISTS (SELECT 1 FROM existing) - ON CONFLICT (did) DO UPDATE SET last_indexed = NOW() RETURNING id, sync_state ) SELECT diff --git a/consumer/src/db/operations/graph.rs b/consumer/src/db/operations/graph.rs index 3c8abb0f..60f44ece 100644 --- a/consumer/src/db/operations/graph.rs +++ b/consumer/src/db/operations/graph.rs @@ -23,23 +23,13 @@ pub async fn follow_insert( conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; - // Insert follow - actors already resolved (no SELECT subquery needed) - // Note: follows table has two unique constraints: - // 1. (actor_id, rkey) - canonical AT Protocol record identifier (record URI) - // 2. (actor_id, subject_actor_id) - business logic (one follow per actor-subject pair) - // - // Strategy: Use (actor_id, subject_actor_id) as primary conflict target with newer-wins - // - Same subject, newer rkey → update to newer - // - Same subject, older/equal rkey → no-op (WHERE clause prevents update) - // - Same rkey, different subject → constraint violation (rare buggy client case - will error) + // Insert follow - simple insert on primary key (actor_id, rkey) // Note: CID is synthetic, generated from actor_id + rkey let rows = conn .execute( "INSERT INTO follows (actor_id, rkey, subject_actor_id) VALUES ($1, $2, $3) - ON CONFLICT (actor_id, subject_actor_id) DO UPDATE SET - rkey = EXCLUDED.rkey - WHERE EXCLUDED.rkey > follows.rkey", + ON CONFLICT (actor_id, rkey) DO NOTHING", &[&actor_id, &rkey, &subject_actor_id], ) .await?; @@ -68,7 +58,7 @@ pub async fn follow_delete( // Block functions -/// Insert a block with CTE to ensure both blocker and blocked actors exist +/// Insert a block with simple upsert on primary key pub async fn block_insert( conn: &C, rkey: i64, @@ -82,23 +72,13 @@ pub async fn block_insert( conn.execute("SELECT pg_advisory_xact_lock($1, $2)", &[&table_id, &key_id]) .await?; - // Insert block - actors already resolved (no SELECT subquery needed) - // Note: blocks table has two unique constraints: - // 1. (actor_id, rkey) - canonical AT Protocol record identifier (record URI) - // 2. (actor_id, subject_actor_id) - business logic (one block per actor-subject pair) - // - // Strategy: Use (actor_id, subject_actor_id) as primary conflict target with newer-wins - // - Same subject, newer rkey → update to newer - // - Same subject, older/equal rkey → no-op (WHERE clause prevents update) - // - Same rkey, different subject → constraint violation (rare buggy client case - will error) + // Insert block - simple insert on primary key (actor_id, rkey) // Note: CID is synthetic, generated from actor_id + rkey let rows = conn .execute( "INSERT INTO blocks (actor_id, rkey, subject_actor_id) VALUES ($1, $2, $3) - ON CONFLICT (actor_id, subject_actor_id) DO UPDATE SET - rkey = EXCLUDED.rkey - WHERE EXCLUDED.rkey > blocks.rkey", + ON CONFLICT (actor_id, rkey) DO NOTHING", &[&actor_id, &rkey, &subject_actor_id], ) .await?; diff --git a/consumer/src/db/workers.rs b/consumer/src/db/workers.rs index 24077c2a..cc7cbc35 100644 --- a/consumer/src/db/workers.rs +++ b/consumer/src/db/workers.rs @@ -21,9 +21,13 @@ pub async fn bulk_ensure_actors(conn: &C, dids: &[&str]) -> Re } conn.execute( - "INSERT INTO actors (did, status, sync_state, last_indexed) - SELECT unnest($1::text[]), 'active', 'partial', NOW() - ON CONFLICT (did) DO NOTHING", + "WITH input_dids AS ( + SELECT unnest($1::text[]) as did + ) + INSERT INTO actors (did, status, sync_state, last_indexed) + SELECT i.did, 'active', 'partial', NOW() + FROM input_dids i + WHERE NOT EXISTS (SELECT 1 FROM actors a WHERE a.did = i.did)", &[&dids], ) .await @@ -62,15 +66,24 @@ pub async fn backfill_update_actor_status( /// * `conn` - Database connection /// * `did` - Actor DID pub async fn backfill_mark_processing(conn: &C, did: &str) -> Result { - conn.execute( - "INSERT INTO actors (did, sync_state, last_indexed) - VALUES ($1, 'processing', NOW()) - ON CONFLICT (did) DO UPDATE - SET sync_state = 'processing', last_indexed=NOW()", + // Try UPDATE first, then INSERT if needed + let updated = conn.execute( + "UPDATE actors SET sync_state = 'processing', last_indexed=NOW() WHERE did=$1", &[&did], ) - .await - .wrap_err_with(|| format!("Failed to mark actor {} as processing", did)) + .await?; + + if updated == 0 { + conn.execute( + "INSERT INTO actors (did, sync_state, last_indexed) + VALUES ($1, 'processing', NOW())", + &[&did], + ) + .await + .wrap_err_with(|| format!("Failed to mark actor {} as processing", did)) + } else { + Ok(updated) + } } /// Get pinned post URI for profile (used for constellation cache warming) diff --git a/consumer/src/workers/backfill/downloader.rs b/consumer/src/workers/backfill/downloader.rs index 098db1cc..53b2516e 100644 --- a/consumer/src/workers/backfill/downloader.rs +++ b/consumer/src/workers/backfill/downloader.rs @@ -125,8 +125,14 @@ pub async fn downloader(config: DownloaderConfig) { ))); } - let status_stmt = conn.prepare_typed_cached( - "INSERT INTO actors (did, sync_state, last_indexed) VALUES ($1, 'processing', NOW()) ON CONFLICT (did) DO UPDATE SET sync_state = 'processing', last_indexed=NOW()", + // Prepare statements for UPDATE-or-INSERT pattern + let update_stmt = conn.prepare_typed_cached( + "UPDATE actors SET sync_state = 'processing', last_indexed=NOW() WHERE did=$1", + &[Type::TEXT] + ).await.unwrap(); + + let insert_stmt = conn.prepare_typed_cached( + "INSERT INTO actors (did, sync_state, last_indexed) VALUES ($1, 'processing', NOW())", &[Type::TEXT] ).await.unwrap(); @@ -237,10 +243,20 @@ pub async fn downloader(config: DownloaderConfig) { } } - // set the repo to processing - if let Err(e) = conn.execute(&status_stmt, &[&did]).await { - tracing::error!(did = %did, error = %e, "Failed to mark actor as processing"); - continue; + // set the repo to processing - try UPDATE first, then INSERT if needed + let updated = match conn.execute(&update_stmt, &[&did]).await { + Ok(rows) => rows, + Err(e) => { + tracing::error!(did = %did, error = %e, "Failed to update actor status"); + continue; + } + }; + + if updated == 0 { + if let Err(e) = conn.execute(&insert_stmt, &[&did]).await { + tracing::error!(did = %did, error = %e, "Failed to insert actor as processing"); + continue; + } } let handle = did_doc diff --git a/migrations/2025-11-20-223826_hypertable_compression/down.sql b/migrations/2025-11-20-223826_hypertable_compression/down.sql new file mode 100644 index 00000000..9644e033 --- /dev/null +++ b/migrations/2025-11-20-223826_hypertable_compression/down.sql @@ -0,0 +1,126 @@ +-- ============================================================================= +-- Revert TimescaleDB Hypertable Conversion +-- ============================================================================= +-- This migration reverts the 8 tables back to regular PostgreSQL tables +-- ============================================================================= + +-- ============================================================================= +-- SECTION 1: Decompress All Chunks (if compressed) +-- ============================================================================= +-- Decompress before copying data to ensure we get all records + +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('postgates') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('threadgates') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('reposts') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('blocks') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('post_aggregate_stats') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('follows') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('actor_aggregate_stats') i; +SELECT decompress_chunk(i, if_compressed => true) FROM show_chunks('actors') i; + +-- ============================================================================= +-- SECTION 2: Create Backup Tables with Data +-- ============================================================================= + +CREATE TABLE postgates_backup AS SELECT * FROM postgates; +CREATE TABLE threadgates_backup AS SELECT * FROM threadgates; +CREATE TABLE reposts_backup AS SELECT * FROM reposts; +CREATE TABLE blocks_backup AS SELECT * FROM blocks; +CREATE TABLE post_aggregate_stats_backup AS SELECT * FROM post_aggregate_stats; +CREATE TABLE follows_backup AS SELECT * FROM follows; +CREATE TABLE actor_aggregate_stats_backup AS SELECT * FROM actor_aggregate_stats; +CREATE TABLE actors_backup AS SELECT * FROM actors; + +-- ============================================================================= +-- SECTION 3: Drop Hypertables (CASCADE removes all chunks) +-- ============================================================================= + +DROP TABLE postgates CASCADE; +DROP TABLE threadgates CASCADE; +DROP TABLE reposts CASCADE; +DROP TABLE blocks CASCADE; +DROP TABLE post_aggregate_stats CASCADE; +DROP TABLE follows CASCADE; +DROP TABLE actor_aggregate_stats CASCADE; +DROP TABLE actors CASCADE; + +-- ============================================================================= +-- SECTION 4: Restore as Regular Tables +-- ============================================================================= + +ALTER TABLE postgates_backup RENAME TO postgates; +ALTER TABLE threadgates_backup RENAME TO threadgates; +ALTER TABLE reposts_backup RENAME TO reposts; +ALTER TABLE blocks_backup RENAME TO blocks; +ALTER TABLE post_aggregate_stats_backup RENAME TO post_aggregate_stats; +ALTER TABLE follows_backup RENAME TO follows; +ALTER TABLE actor_aggregate_stats_backup RENAME TO actor_aggregate_stats; +ALTER TABLE actors_backup RENAME TO actors; + +-- ============================================================================= +-- SECTION 5: Restore Primary Keys and Constraints +-- ============================================================================= + +-- Restore primary keys first (required before FK constraints) +ALTER TABLE actors ADD PRIMARY KEY (id); +ALTER TABLE postgates ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE threadgates ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE reposts ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE blocks ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE post_aggregate_stats ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE follows ADD PRIMARY KEY (actor_id, rkey); +ALTER TABLE actor_aggregate_stats ADD PRIMARY KEY (actor_id); + +-- Restore unique constraints +ALTER TABLE reposts + ADD CONSTRAINT reposts_actor_id_post_actor_id_post_rkey_key + UNIQUE (actor_id, post_actor_id, post_rkey); + +ALTER TABLE blocks + ADD CONSTRAINT blocks_actor_id_subject_actor_id_key + UNIQUE (actor_id, subject_actor_id); + +ALTER TABLE follows + ADD CONSTRAINT follows_actor_id_subject_actor_id_key + UNIQUE (actor_id, subject_actor_id); + +ALTER TABLE actors + ADD CONSTRAINT actors_did_key + UNIQUE (did); + +-- Restore FK constraints +ALTER TABLE postgates + ADD CONSTRAINT postgates_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE threadgates + ADD CONSTRAINT threadgates_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE reposts + ADD CONSTRAINT reposts_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE blocks + ADD CONSTRAINT blocks_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE blocks + ADD CONSTRAINT blocks_subject_actor_id_fkey + FOREIGN KEY (subject_actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE follows + ADD CONSTRAINT follows_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE follows + ADD CONSTRAINT follows_subject_actor_id_fkey + FOREIGN KEY (subject_actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +ALTER TABLE actor_aggregate_stats + ADD CONSTRAINT actor_aggregate_stats_actor_id_fkey + FOREIGN KEY (actor_id) REFERENCES actors(id) ON DELETE CASCADE; + +-- ============================================================================= +-- Note: Indexes will be automatically recreated by Diesel based on schema.rs +-- ============================================================================= diff --git a/migrations/2025-11-20-223826_hypertable_compression/up.sql b/migrations/2025-11-20-223826_hypertable_compression/up.sql new file mode 100644 index 00000000..2a33d338 --- /dev/null +++ b/migrations/2025-11-20-223826_hypertable_compression/up.sql @@ -0,0 +1,274 @@ +-- ============================================================================= +-- TimescaleDB Hypertable Conversion & Compression +-- ============================================================================= +-- +-- Converts 8 large tables to TimescaleDB hypertables with compression: +-- - postgates (1.3 MB) +-- - threadgates (4.2 MB) +-- - reposts (17 MB) +-- - blocks (17 MB) +-- - post_aggregate_stats (30 MB) +-- - follows (44 MB) +-- - actor_aggregate_stats (25 MB) +-- - actors (55 MB) +-- +-- Total size before: 193 MB +-- Expected size after: 45-70 MB (64-77% reduction) +-- +-- NOTE: TimescaleDB does not support foreign keys between hypertables, so we +-- drop FK constraints where both source and target are becoming hypertables. +-- Referential integrity is enforced at the application level. +-- ============================================================================= + +-- ============================================================================= +-- SECTION 1: Drop Constraints +-- ============================================================================= + +-- Drop ALL FK constraints pointing to actors table (42 constraints) +-- TimescaleDB does not support FK constraints pointing TO hypertables +-- Application must enforce referential integrity instead +ALTER TABLE actor_aggregate_stats DROP CONSTRAINT IF EXISTS actor_aggregate_stats_actor_id_fkey; +ALTER TABLE blocks DROP CONSTRAINT IF EXISTS blocks_actor_id_fkey; +ALTER TABLE blocks DROP CONSTRAINT IF EXISTS blocks_subject_actor_id_fkey; +ALTER TABLE bookmarks DROP CONSTRAINT IF EXISTS bookmarks_actor_id_fkey; +ALTER TABLE chat_decls DROP CONSTRAINT IF EXISTS chat_decls_actor_id_fkey; +ALTER TABLE feedgen_likes DROP CONSTRAINT IF EXISTS feedgen_likes_actor_id_fkey; +ALTER TABLE feedgens DROP CONSTRAINT IF EXISTS feedgens_owner_actor_id_fkey; +ALTER TABLE feedgens DROP CONSTRAINT IF EXISTS feedgens_service_actor_id_fkey; +ALTER TABLE feedgens DROP CONSTRAINT IF EXISTS feedgens_actor_id_fkey; +ALTER TABLE follows DROP CONSTRAINT IF EXISTS follows_actor_id_fkey; +ALTER TABLE follows DROP CONSTRAINT IF EXISTS follows_subject_actor_id_fkey; +ALTER TABLE labeler_defs DROP CONSTRAINT IF EXISTS labeler_defs_labeler_actor_id_fkey; +ALTER TABLE labeler_likes DROP CONSTRAINT IF EXISTS labeler_likes_labeler_actor_id_fkey; +ALTER TABLE labeler_likes DROP CONSTRAINT IF EXISTS labeler_likes_actor_id_fkey; +ALTER TABLE labelers DROP CONSTRAINT IF EXISTS labelers_actor_id_fkey; +ALTER TABLE labels DROP CONSTRAINT IF EXISTS labels_labeler_actor_id_fkey; +ALTER TABLE list_blocks DROP CONSTRAINT IF EXISTS list_blocks_actor_id_fkey; +ALTER TABLE list_items DROP CONSTRAINT IF EXISTS list_items_subject_actor_id_fkey; +ALTER TABLE list_items DROP CONSTRAINT IF EXISTS list_items_actor_id_fkey; +ALTER TABLE list_mutes DROP CONSTRAINT IF EXISTS list_mutes_actor_id_fkey; +ALTER TABLE lists DROP CONSTRAINT IF EXISTS lists_owner_actor_id_fkey; +ALTER TABLE lists DROP CONSTRAINT IF EXISTS lists_actor_id_fkey; +ALTER TABLE mutes DROP CONSTRAINT IF EXISTS mutes_actor_id_fkey; +ALTER TABLE mutes DROP CONSTRAINT IF EXISTS mutes_subject_actor_id_fkey; +ALTER TABLE notif_decl DROP CONSTRAINT IF EXISTS notif_decl_actor_id_fkey; +ALTER TABLE notification_state DROP CONSTRAINT IF EXISTS notification_state_actor_id_fkey; +ALTER TABLE notifications DROP CONSTRAINT IF EXISTS notifications_recipient_actor_id_fkey; +ALTER TABLE notifications DROP CONSTRAINT IF EXISTS notifications_author_actor_id_fkey; +ALTER TABLE notifications DROP CONSTRAINT IF EXISTS notifications_subject_actor_id_fkey; +ALTER TABLE post_likes DROP CONSTRAINT IF EXISTS post_likes_actor_id_fkey; +ALTER TABLE postgates DROP CONSTRAINT IF EXISTS postgates_actor_id_fkey; +ALTER TABLE posts DROP CONSTRAINT IF EXISTS posts_actor_id_fkey; +ALTER TABLE profiles DROP CONSTRAINT IF EXISTS profiles_actor_id_fkey; +ALTER TABLE reposts DROP CONSTRAINT IF EXISTS reposts_actor_id_fkey; +ALTER TABLE starterpacks DROP CONSTRAINT IF EXISTS starterpacks_actor_id_fkey; +ALTER TABLE starterpacks DROP CONSTRAINT IF EXISTS starterpacks_owner_actor_id_fkey; +ALTER TABLE statuses DROP CONSTRAINT IF EXISTS statuses_actor_id_fkey; +ALTER TABLE thread_mutes DROP CONSTRAINT IF EXISTS thread_mutes_actor_id_fkey; +ALTER TABLE threadgates DROP CONSTRAINT IF EXISTS threadgates_actor_id_fkey; +ALTER TABLE verification DROP CONSTRAINT IF EXISTS verification_actor_id_fkey; +ALTER TABLE verification DROP CONSTRAINT IF EXISTS verification_verifier_actor_id_fkey; +ALTER TABLE verification DROP CONSTRAINT IF EXISTS verification_subject_actor_id_fkey; + +-- Drop unique constraints that don't include partition column +-- (TimescaleDB requires unique indexes to include partitioning dimension) +-- Application must enforce these business rules instead +ALTER TABLE reposts DROP CONSTRAINT IF EXISTS reposts_actor_id_post_actor_id_post_rkey_key; +ALTER TABLE blocks DROP CONSTRAINT IF EXISTS blocks_actor_id_subject_actor_id_key; +ALTER TABLE follows DROP CONSTRAINT IF EXISTS follows_actor_id_subject_actor_id_key; +ALTER TABLE actors DROP CONSTRAINT IF EXISTS actors_did_key; + +-- ============================================================================= +-- SECTION 2: Convert actors Table FIRST +-- ============================================================================= +-- Convert actors before other tables to avoid FK constraint issues + +-- --------------------------------------------------------------------------- +-- actors (55 MB) - Core actor metadata +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'actors', + by_range('id', 100000), -- 100k actor IDs per chunk + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE actors SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'status', + timescaledb.compress_orderby = 'id DESC' +); + +SELECT enable_chunk_skipping('actors', 'id'); + +COMMENT ON TABLE actors IS 'TimescaleDB hypertable for actor metadata. Partitioned by id with 100k chunks. Manual compression.'; + +-- ============================================================================= +-- SECTION 3: Convert rkey-Partitioned Tables to Hypertables +-- ============================================================================= +-- These tables use rkey (TID timestamp) for partitioning with 7-day chunks +-- TID encoding: 7 days = 604800 seconds = 604800000000 microseconds << 10 bits = 619315200000000 + +-- --------------------------------------------------------------------------- +-- postgates (1.3 MB) - Post visibility rules +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'postgates', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE postgates SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('postgates', 'actor_id'); +SELECT enable_chunk_skipping('postgates', 'post_actor_id'); + +COMMENT ON TABLE postgates IS 'TimescaleDB hypertable for post visibility rules. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- --------------------------------------------------------------------------- +-- threadgates (4.2 MB) - Thread reply restrictions +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'threadgates', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE threadgates SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('threadgates', 'actor_id'); +SELECT enable_chunk_skipping('threadgates', 'post_actor_id'); + +COMMENT ON TABLE threadgates IS 'TimescaleDB hypertable for thread reply restrictions. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- --------------------------------------------------------------------------- +-- reposts (17 MB) - Repost records +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'reposts', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE reposts SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('reposts', 'actor_id'); +SELECT enable_chunk_skipping('reposts', 'post_actor_id'); + +COMMENT ON TABLE reposts IS 'TimescaleDB hypertable for reposts. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- --------------------------------------------------------------------------- +-- blocks (17 MB) - Block relationships +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'blocks', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE blocks SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('blocks', 'actor_id'); +SELECT enable_chunk_skipping('blocks', 'subject_actor_id'); + +COMMENT ON TABLE blocks IS 'TimescaleDB hypertable for block relationships. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- --------------------------------------------------------------------------- +-- post_aggregate_stats (30 MB) - Cached post statistics +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'post_aggregate_stats', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE post_aggregate_stats SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('post_aggregate_stats', 'actor_id'); + +COMMENT ON TABLE post_aggregate_stats IS 'TimescaleDB hypertable for cached post statistics. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- --------------------------------------------------------------------------- +-- follows (44 MB) - Social graph relationships +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'follows', + by_range('rkey', 619315200000000), -- 7 days in TID units + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE follows SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'actor_id', + timescaledb.compress_orderby = 'rkey DESC' +); + +SELECT enable_chunk_skipping('follows', 'actor_id'); +SELECT enable_chunk_skipping('follows', 'subject_actor_id'); + +COMMENT ON TABLE follows IS 'TimescaleDB hypertable for follow relationships. Partitioned by rkey with 7-day chunks. Manual compression.'; + +-- ============================================================================= +-- SECTION 4: Convert actor_aggregate_stats Table +-- ============================================================================= +-- This table uses actor_id (integer) for partitioning with 100k chunks + +-- --------------------------------------------------------------------------- +-- actor_aggregate_stats (25 MB) - Cached actor statistics +-- --------------------------------------------------------------------------- +SELECT create_hypertable( + 'actor_aggregate_stats', + by_range('actor_id', 100000), -- 100k actor IDs per chunk + migrate_data => true, + if_not_exists => true +); + +ALTER TABLE actor_aggregate_stats SET ( + timescaledb.compress, + timescaledb.compress_segmentby = 'stat_type', + timescaledb.compress_orderby = 'actor_id DESC' +); + +SELECT enable_chunk_skipping('actor_aggregate_stats', 'actor_id'); + +COMMENT ON TABLE actor_aggregate_stats IS 'TimescaleDB hypertable for cached actor statistics. Partitioned by actor_id with 100k chunks. Manual compression.'; + +-- ============================================================================= +-- SECTION 5: Compress All Chunks +-- ============================================================================= +-- Manually compress all chunks for all converted tables + +SELECT compress_chunk(i) FROM show_chunks('postgates') i; +SELECT compress_chunk(i) FROM show_chunks('threadgates') i; +SELECT compress_chunk(i) FROM show_chunks('reposts') i; +SELECT compress_chunk(i) FROM show_chunks('blocks') i; +SELECT compress_chunk(i) FROM show_chunks('post_aggregate_stats') i; +SELECT compress_chunk(i) FROM show_chunks('follows') i; +SELECT compress_chunk(i) FROM show_chunks('actor_aggregate_stats') i; +SELECT compress_chunk(i) FROM show_chunks('actors') i; diff --git a/parakeet-db/src/schema.rs b/parakeet-db/src/schema.rs index 719361f7..15123698 100644 --- a/parakeet-db/src/schema.rs +++ b/parakeet-db/src/schema.rs @@ -718,33 +718,14 @@ diesel::table! { } } -diesel::joinable!(actor_aggregate_stats -> actors (actor_id)); -diesel::joinable!(bookmarks -> actors (actor_id)); -diesel::joinable!(chat_decls -> actors (actor_id)); -diesel::joinable!(feedgen_likes -> actors (actor_id)); -diesel::joinable!(labeler_defs -> actors (labeler_actor_id)); -diesel::joinable!(labelers -> actors (actor_id)); -diesel::joinable!(labels -> actors (labeler_actor_id)); -diesel::joinable!(list_blocks -> actors (actor_id)); diesel::joinable!(list_blocks -> lists (list_id)); diesel::joinable!(list_items -> lists (list_id)); -diesel::joinable!(list_mutes -> actors (actor_id)); diesel::joinable!(list_mutes -> lists (list_id)); -diesel::joinable!(notif_decl -> actors (actor_id)); -diesel::joinable!(notification_state -> actors (actor_id)); -diesel::joinable!(post_likes -> actors (actor_id)); -diesel::joinable!(postgates -> actors (actor_id)); -diesel::joinable!(posts -> actors (actor_id)); -diesel::joinable!(profiles -> actors (actor_id)); diesel::joinable!(profiles -> starterpacks (joined_sp_id)); -diesel::joinable!(reposts -> actors (actor_id)); diesel::joinable!(starterpack_feeds -> feedgens (feed_id)); diesel::joinable!(starterpack_feeds -> starterpacks (starterpack_id)); diesel::joinable!(starterpacks -> lists (list_id)); -diesel::joinable!(statuses -> actors (actor_id)); -diesel::joinable!(thread_mutes -> actors (actor_id)); diesel::joinable!(threadgate_allowed_lists -> lists (list_id)); -diesel::joinable!(threadgates -> actors (actor_id)); diesel::allow_tables_to_appear_in_same_query!( _diesel_schema_inference, -- 2.51.2