store if a post violates a threadgate

consumer/src/db/copy.rs

···
 
     writer.finish().await?;
 
+    let threadgated: Vec<(String, String, DateTime<Utc>)> = conn
+        .query(
+            "SELECT root_uri, p.at_uri, p.created_at FROM posts_tmp p INNER JOIN threadgates t ON root_uri = post_uri WHERE t.allow IS NOT NULL",
+            &[],
+        )
+        .await?
+        .into_iter()
+        .map(|v| (v.get(0), v.get(1), v.get(2))).collect();
+
+    for (root, post, created_at) in dbg!(threadgated) {
+        match super::post_enforce_threadgate(conn, &root, did, created_at, true).await {
+            Ok(true) => {
+                conn.execute(
+                    "UPDATE posts_tmp SET violates_threadgate=TRUE WHERE at_uri=$1",
+                    &[&post],
+                )
+                .await?;
+            }
+            Ok(false) => continue,
+            Err(e) => {
+                tracing::error!("failed to check threadgate enforcement: {e}");
+                continue;
+            }
+        }
+    }
+
     conn.execute("INSERT INTO posts (SELECT * FROM posts_tmp)", &[])
         .await
 }

consumer/src/db/record.rs

···
     let embed = rec.embed.as_ref().map(|v| v.as_str());
     let embed_subtype = rec.embed.as_ref().and_then(|v| v.subtype());
 
+    // if there is a root, we need to check for the presence of a threadgate.
+    let violates_threadgate = match &root_uri {
+        Some(root) => {
+            post_enforce_threadgate(conn, root, repo, rec.created_at, is_backfill).await?
+        }
+        None => false,
+    };
+
     let tags = merge_tags(tags, rec.tags);
 
     let count = conn
···
                 &embed,
                 &embed_subtype,
                 &mentions,
+                &violates_threadgate,
                 &rec.created_at,
             ],
         )
···
 pub async fn post_delete<C: GenericClient>(conn: &mut C, at_uri: &str) -> PgExecResult {
     conn.execute("DELETE FROM posts WHERE at_uri=$1", &[&at_uri])
         .await
+}
+
+pub async fn post_enforce_threadgate<C: GenericClient>(
+    conn: &mut C,
+    root: &str,
+    post_author: &str,
+    post_created_at: DateTime<Utc>,
+    is_backfill: bool,
+) -> PgResult<bool> {
+    // check if the root and the current post are the same author
+    // strip "at://" then break into parts by '/'
+    let parts = root[5..].split('/').collect::<Vec<_>>();
+    let root_author = parts[0];
+    if root_author == post_author {
+        return Ok(false);
+    }
+
+    let tg_data = threadgate_get(conn, root).await?;
+
+    let Some((created_at, allow, allow_lists)) = tg_data else {
+        return Ok(false);
+    };
+
+    // when backfilling, there's no point continuing if the record is dated before the threadgate
+    if is_backfill && post_created_at < created_at {
+        return Ok(false);
+    }
+
+    if allow.is_empty() {
+        return Ok(true);
+    }
+
+    let allow: HashSet<String> = HashSet::from_iter(allow);
+
+    if allow.contains("app.bsky.feed.threadgate#followerRule")
+        || allow.contains("app.bsky.feed.threadgate#followingRule")
+    {
+        let profile_state: Option<(bool, bool)> = conn
+            .query_opt(
+                "SELECT following IS NOT NULL, followed IS NOT NULL FROM profile_states WHERE did=$1 AND subject=$2",
+                &[&root_author, &post_author],
+            )
+            .await?
+            .map(|v| (v.get(0), v.get(1)));
+
+        if let Some((following, followed)) = profile_state {
+            if allow.contains("app.bsky.feed.threadgate#followerRule") && followed {
+                return Ok(false);
+            }
+
+            if allow.contains("app.bsky.feed.threadgate#followingRule") && following {
+                return Ok(false);
+            }
+        }
+    }
+
+    // check mentions
+    if allow.contains("app.bsky.feed.threadgate#mentionRule") {
+        let mentions: Vec<String> = conn
+            .query_opt("SELECT mentions FROM posts WHERE at_uri=$1", &[&root])
+            .await?
+            .map(|r| r.get(0))
+            .unwrap_or_default();
+
+        if mentions.contains(&post_author.to_owned()) {
+            return Ok(false);
+        }
+    }
+
+    if allow.contains("app.bsky.feed.threadgate#listRule") {
+        if allow_lists.is_empty() {
+            return Ok(true);
+        }
+
+        let count: i64 = conn
+            .query_one(
+                "SELECT count(*) FROM list_items WHERE list_uri=ANY($1) AND subject=$2",
+                &[&allow_lists, &post_author],
+            )
+            .await?
+            .get(0);
+        if count == 0 {
+            return Ok(true);
+        }
+    }
+
+    Ok(false)
 }
 
 pub async fn post_get_info_for_delete<C: GenericClient>(
···
 pub async fn status_delete<C: GenericClient>(conn: &mut C, did: &str) -> PgExecResult {
     conn.execute("DELETE FROM statuses WHERE did=$1", &[&did])
         .await
+}
+
+async fn threadgate_get<C: GenericClient>(
+    conn: &mut C,
+    post: &str,
+) -> PgOptResult<(DateTime<Utc>, Vec<String>, Vec<String>)> {
+    let res = conn
+        .query_opt(
+            "SELECT created_at, allow, allowed_lists FROM threadgates WHERE post_uri=$1 AND allow IS NOT NULL",
+            &[&post],
+        )
+        .await?
+        .map(|v| (v.get(0), v.get(1), v.get(2)));
+
+    Ok(res)
 }
 
 pub async fn threadgate_upsert<C: GenericClient>(

consumer/src/db/sql/post_insert.sql

···
 INSERT INTO posts (at_uri, did, cid, record, content, facets, languages, tags, parent_uri, parent_cid, root_uri,
-                   root_cid, embed, embed_subtype, mentions, created_at)
-VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
+                   root_cid, embed, embed_subtype, mentions, violates_threadgate, created_at)
+VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
 ON CONFLICT DO NOTHING

migrations/2025-09-27-171241_post-tweaks/down.sql

···
 alter table posts
-    drop column mentions;
+    drop column mentions,
+    drop column violates_threadgate;

migrations/2025-09-27-171241_post-tweaks/up.sql

···
 alter table posts
-    add column mentions text[];
+    add column mentions            text[],
+    add column violates_threadgate bool not null default false;

parakeet-db/src/schema.rs

···
         created_at -> Timestamptz,
         indexed_at -> Timestamp,
         mentions -> Nullable<Array<Nullable<Text>>>,
+        violates_threadgate -> Bool,
     }
 }