feat: add account_created_at

Cargo.lock

···
 name = "did-resolver"
 version = "0.1.0"
 dependencies = [
+ "chrono",
  "hickory-resolver",
  "reqwest",
  "serde",

consumer/src/database_writer/operations/executor.rs

···
             status,
             sync_state,
             handle,
+            account_created_at,
             timestamp,
         } => {
             db::actor_upsert(
···
                 status.as_ref(),
                 &sync_state,
                 handle.as_deref(),
+                account_created_at.as_ref(),
                 timestamp,
             )
             .await?;

consumer/src/database_writer/operations/types.rs

···
         status: Option<parakeet_db::types::ActorStatus>,
         sync_state: parakeet_db::types::ActorSyncState,
         handle: Option<String>,
+        account_created_at: Option<DateTime<Utc>>,
         timestamp: DateTime<Utc>,
     },
 

consumer/src/db/actor.rs

···
     status: Option<&ActorStatus>,
     sync_state: &ActorSyncState,
     handle: Option<&str>,
+    account_created_at: Option<&DateTime<Utc>>,
     time: DateTime<Utc>,
 ) -> Result<u64> {
     // Allow allowlist states (synced, dirty, processing) to flow freely
     // Allow upgrading from partial to allowlist states
     // Never downgrade from allowlist states to partial
+    //
+    // Account created_at is updated if provided and current value is NULL
+    // This allows enrichment during handle resolution without overwriting existing values
 
-    match (status, handle) {
-        (Some(status), Some(handle)) => {
-            // Both status and handle provided
+    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,
+                    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,
+                    account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at),
+                    last_indexed=EXCLUDED.last_indexed",
+                &[&did, &status, &handle, &sync_state, &created_at, &time],
+            )
+            .await
+        }
+        (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
···
             )
             .await
         }
-        (Some(status), None) => {
+        (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,
+                    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,
+                    account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at),
+                    last_indexed=EXCLUDED.last_indexed",
+                &[&did, &status, &sync_state, &created_at, &time],
+            )
+            .await
+        }
+        (Some(status), None, None) => {
             // Only status provided
             conn.execute(
                 "INSERT INTO actors (did, status, sync_state, last_indexed) VALUES ($1, $2, $3, $4)
···
             )
             .await
         }
-        (None, Some(handle)) => {
+        (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,
+                    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,
+                    account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at),
+                    last_indexed=EXCLUDED.last_indexed",
+                &[&did, &handle, &sync_state, &created_at, &time],
+            )
+            .await
+        }
+        (None, Some(handle), None) => {
             // Only handle provided
             conn.execute(
                 "INSERT INTO actors (did, handle, sync_state, last_indexed) VALUES ($1, $2, $3, $4)
···
             )
             .await
         }
-        (None, None) => {
+        (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
+                    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,
+                    account_created_at=COALESCE(actors.account_created_at, EXCLUDED.account_created_at),
+                    last_indexed=EXCLUDED.last_indexed",
+                &[&did, &sync_state, &created_at, &time],
+            )
+            .await
+        }
+        (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)

consumer/src/db/allowlist.rs

···
             Some(&ActorStatus::Active),
             &ActorSyncState::Dirty,
             None, // no handle
+            None, // no account_created_at (enriched during handle resolution)
             now,
         )
         .await

consumer/src/workers/handle_resolver.rs

···
         let cache_key = format!("did_handle:{}", did);
         let cached_handle: Option<String> = self.redis.get(&cache_key).await?;
 
-        let handle = if let Some(cached) = cached_handle {
+        let (handle, account_created_at) = if let Some(cached) = cached_handle {
             counter!("handle_resolver.cache_hit").increment(1);
-            Some(cached)
+            // Cache hit - we have the handle but not the account creation time
+            // Fetch account creation time separately (not cached, but only fetched once per actor typically)
+            let created_at = match self.resolver.get_plc_creation_time(did).await {
+                Ok(Some(timestamp)) => {
+                    counter!("handle_resolver.plc_creation_fetch_success").increment(1);
+                    Some(timestamp)
+                }
+                Ok(None) => {
+                    debug!("No PLC creation time found for {}", did);
+                    counter!("handle_resolver.plc_creation_not_found").increment(1);
+                    None
+                }
+                Err(e) => {
+                    debug!("Failed to fetch PLC creation time for {}: {}", did, e);
+                    counter!("handle_resolver.plc_creation_error").increment(1);
+                    None
+                }
+            };
+            (Some(cached), created_at)
         } else {
             // 2. Cache miss - resolve from PLC directory
             counter!("handle_resolver.cache_miss").increment(1);
 
-            match self.resolver.resolve_did(did).await {
+            let (handle, created_at) = match self.resolver.resolve_did(did).await {
                 Ok(Some(doc)) => {
                     // Extract handle from DID document's alsoKnownAs field
                     let handle = doc
···
                         counter!("handle_resolver.no_handle").increment(1);
                     }
 
-                    handle
+                    // Fetch account creation time from PLC audit log
+                    let created_at = match self.resolver.get_plc_creation_time(did).await {
+                        Ok(Some(timestamp)) => {
+                            counter!("handle_resolver.plc_creation_fetch_success").increment(1);
+                            Some(timestamp)
+                        }
+                        Ok(None) => {
+                            debug!("No PLC creation time found for {}", did);
+                            counter!("handle_resolver.plc_creation_not_found").increment(1);
+                            None
+                        }
+                        Err(e) => {
+                            debug!("Failed to fetch PLC creation time for {}: {}", did, e);
+                            counter!("handle_resolver.plc_creation_error").increment(1);
+                            None
+                        }
+                    };
+
+                    (handle, created_at)
                 }
                 Ok(None) => {
                     debug!("DID document not found for {}", did);
                     counter!("handle_resolver.did_not_found").increment(1);
-                    None
+                    (None, None)
                 }
                 Err(e) => {
                     debug!("Failed to resolve DID for {}: {}", did, e);
                     counter!("handle_resolver.did_resolution_error").increment(1);
-                    None
+                    (None, None)
                 }
-            }
+            };
+
+            (handle, created_at)
         };
 
         // 4. Send handle update operation to batch writer (bounded channel with backpressure)
···
                 status: None,
                 sync_state: parakeet_db::types::ActorSyncState::Partial,
                 handle,
+                account_created_at,
                 timestamp: chrono::Utc::now(),
             }],
             cache_invalidations: vec![],

consumer/src/workers/jetstream/processing.rs

···
         status: None,
         sync_state,
         handle, // Pass the Option<String> directly
+        account_created_at: None, // Jetstream doesn't provide creation time, enriched during handle resolution
         timestamp,
     }];
 
···
         status: Some(status),
         sync_state,
         handle: None,
+        account_created_at: None, // Jetstream doesn't provide creation time, enriched during handle resolution
         timestamp,
     }];
 

did-resolver/Cargo.toml

···
 edition = "2021"
 
 [dependencies]
+chrono = { version = "0.4", features = ["serde"] }
 hickory-resolver = "0.24.2"
 reqwest = { version = "0.12.12", features = ["json", "native-tls"] }
 serde = { version = "1.0.217", features = ["derive"] }

did-resolver/src/lib.rs

···
         Ok(Some(did_doc))
     }
 
+    /// Get account creation timestamp from PLC audit log
+    /// Returns the timestamp of the first operation (where prev is null)
+    /// Only works for did:plc DIDs - returns None for other DID methods
+    pub async fn get_plc_creation_time(
+        &self,
+        did: &str,
+    ) -> Result<Option<chrono::DateTime<chrono::Utc>>, Error> {
+        // Only fetch for did:plc
+        if !did.starts_with("did:plc:") {
+            return Ok(None);
+        }
+
+        let res = self
+            .client
+            .get(format!("{}/{did}/log/audit", self.plc))
+            .send()
+            .await?;
+
+        let status = res.status();
+
+        if status.is_server_error() {
+            return Err(Error::ServerError);
+        }
+
+        if status == StatusCode::NOT_FOUND || status == StatusCode::GONE {
+            return Ok(None);
+        }
+
+        let audit_log: Vec<types::PlcAuditLogEntry> = res.json().await?;
+
+        // First entry in the audit log is the account creation
+        Ok(audit_log.first().map(|entry| entry.created_at))
+    }
+
     async fn resolve_did_web(&self, id: &str) -> Result<Option<types::DidDocument>, Error> {
         let res = match self
             .client

did-resolver/src/types.rs

···
 
 #[derive(Debug, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
+pub struct PlcAuditLogEntry {
+    pub did: String,
+    pub created_at: chrono::DateTime<chrono::Utc>,
+    // Other fields exist but we only need createdAt
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
 pub struct DidDocument {
     #[serde(default, rename = "@context")]
     pub context: Vec<String>,

migrations/2025-11-05-120000_add_account_created_at_to_actors/down.sql

···
+-- Remove account_created_at column from actors table
+DROP INDEX IF EXISTS idx_actors_account_created_at;
+ALTER TABLE actors DROP COLUMN account_created_at;

migrations/2025-11-05-120000_add_account_created_at_to_actors/up.sql

···
+-- Add account_created_at column to actors table
+-- This represents when the account was created (from PLC directory first operation)
+-- Nullable because we may not know it for all actors initially (especially stubs)
+-- Will be populated during handle resolution
+ALTER TABLE actors ADD COLUMN account_created_at TIMESTAMP WITH TIME ZONE;
+
+-- Create index for potential queries by account age
+CREATE INDEX idx_actors_account_created_at ON actors(account_created_at) WHERE account_created_at IS NOT NULL;

parakeet-db/src/schema.rs

···
         repo_rev -> Nullable<Text>,
         repo_cid -> Nullable<Bytea>,
         last_indexed -> Nullable<Timestamptz>,
+        account_created_at -> Nullable<Timestamptz>,
     }
 }
 

parakeet/src/hydration/profile/builders.rs

···
 }
 
 pub(super) fn build_basic(
-    (did, handle, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
+    (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
     stats: Option<ProfileStats>,
     labels: Vec<models::Label>,
     verifications: Option<Vec<crate::loaders::EnrichedVerification>>,
···
         })
         .unwrap_or_else(|| (None, None, None));
 
-    // Use profile.created_at if available, fall back to Utc::now()
-    let created_at = profile.as_ref().map(|p| p.created_at).unwrap_or_else(Utc::now);
+    // Use actor.account_created_at if available (from PLC directory), fall back to Utc::now()
+    // This is the actual account creation time, not when the profile record was created
+    let created_at = account_created_at.unwrap_or_else(Utc::now);
 
     ProfileViewBasic {
         did,
···
 }
 
 pub(super) fn build_profile(
-    (did, handle, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
+    (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
     stats: Option<ProfileStats>,
     labels: Vec<models::Label>,
     verifications: Option<Vec<crate::loaders::EnrichedVerification>>,
···
         })
         .unwrap_or_else(|| (None, None, None, None));
 
-    // Use profile.created_at for both createdAt and indexedAt (if profile exists)
-    // Fall back to Utc::now() if no profile record exists yet
-    let created_at = profile.as_ref().map(|p| p.created_at).unwrap_or_else(Utc::now);
+    // Use actor.account_created_at for both createdAt and indexedAt
+    // Fall back to Utc::now() if account creation time not available
+    let created_at = account_created_at.unwrap_or_else(Utc::now);
 
     ProfileView {
         did,
···
 }
 
 pub(super) fn build_detailed(
-    (did, handle, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
+    (did, handle, account_created_at, profile, chat_decl, is_labeler, status, notif_decl): ProfileLoaderRet,
     stats: Option<ProfileStats>,
     labels: Vec<models::Label>,
     verifications: Option<Vec<crate::loaders::EnrichedVerification>>,
···
         })
         .unwrap_or_else(|| (None, None, None, None, None, None));
 
-    // Use profile.created_at for both createdAt and indexedAt (if profile exists)
-    // Fall back to Utc::now() if no profile record exists yet
-    let created_at = profile.as_ref().map(|p| p.created_at).unwrap_or_else(Utc::now);
+    // Use actor.account_created_at for both createdAt and indexedAt
+    // Fall back to Utc::now() if account creation time not available
+    let created_at = account_created_at.unwrap_or_else(Utc::now);
 
     ProfileViewDetailed {
         did,

parakeet/src/loaders/profile.rs

···
         "SELECT
             a.did,
             a.handle,
+            a.account_created_at,
             p.actor_id,
             p.cid,
             p.created_at,
···
 
 pub struct ProfileLoader(pub(super) Pool<AsyncPgConnection>, pub(super) redis::aio::MultiplexedConnection);
 pub type ProfileLoaderRet = (
-    String,                   // did - always present (from actors table)
-    Option<String>,           // handle - from actors table
-    Option<models::Profile>,  // profile - optional (may not exist)
+    String,                                      // did - always present (from actors table)
+    Option<String>,                              // handle - from actors table
+    Option<chrono::DateTime<chrono::Utc>>,       // account_created_at - from actors table
+    Option<models::Profile>,                     // profile - optional (may not exist)
     Option<ChatAllowIncoming>,
     bool,
     Option<EnrichedStatus>,
···
             did: String,
             #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Text>)]
             handle: Option<String>,
+            #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Timestamptz>)]
+            account_created_at: Option<chrono::DateTime<chrono::Utc>>,
             #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Integer>)]
             actor_id: Option<i32>,
             #[diesel(sql_type = diesel::sql_types::Nullable<diesel::sql_types::Binary>)]
···
                         let is_labeler = row.labeler_actor_id.is_some();
                         let status = status_map.get(&row.did).cloned();
 
-                        let val = (row.did.clone(), row.handle, profile, chat_decl, is_labeler, status, notif_decl);
+                        let val = (row.did.clone(), row.handle, row.account_created_at, profile, chat_decl, is_labeler, status, notif_decl);
 
                         (row.did, val)
                     },
···
                 // Enqueue missing profiles for background fetching using Redis
                 let missing_profiles: Vec<String> = results
                     .iter()
-                    .filter(|(_, (_, _, profile, _, _, _, _))| profile.is_none())
+                    .filter(|(_, (_, _, _, profile, _, _, _, _))| profile.is_none())
                     .map(|(did, _)| format!("at://{}/app.bsky.actor.profile/self", did))
                     .collect();
 

parakeet/src/xrpc/app_bsky/actor.rs

···
                     // Check if has profile (display name or description)
                     profiles
                         .get(did)
-                        .and_then(|p| p.2.as_ref()) // .2 is Option<models::Profile>
+                        .and_then(|p| p.3.as_ref()) // .3 is Option<models::Profile>
                         .map(|prof| {
                             prof.display_name.is_some() || prof.description.is_some()
                         })

parakeet/src/xrpc/app_bsky/notification/mod.rs

···
         let profile_state = profile_states.get(&notif.author_did);
 
         // Build associated object
-        let associated = if let Some((_, _, _, chat_decl, _, _, notif_decl)) = profile_data {
+        let associated = if let Some((_, _, _, _, chat_decl, _, _, notif_decl)) = profile_data {
             let mut assoc = serde_json::Map::new();
 
             if let Some(chat) = chat_decl {
···
         let author = NotificationAuthor {
             did: notif.author_did.clone(),
             handle: profile_data
-                .and_then(|(_, h, _, _, _, _, _)| h.clone())
+                .and_then(|(_, h, _, _, _, _, _, _)| h.clone())
                 .unwrap_or_else(|| {
                     // Fallback to handle.invalid if profile not yet indexed
                     "handle.invalid".to_string()
                 }),
-            display_name: profile_data.and_then(|(_, _, p, _, _, _, _)| {
+            display_name: profile_data.and_then(|(_, _, _, p, _, _, _, _)| {
                 p.as_ref().and_then(|prof| prof.display_name.clone())
             }),
-            avatar: profile_data.and_then(|(did, _, p, _, _, _, _)| {
+            avatar: profile_data.and_then(|(did, _, _, p, _, _, _, _)| {
                 p.as_ref().and_then(|prof| {
                     prof.avatar_cid.as_ref().and_then(|cid_digest| {
                         parakeet_db::cid_util::digest_to_blob_cid_string(cid_digest)
···
             associated,
             viewer,
             labels: Some(Vec::new()), // Empty labels array for compatibility
-            description: profile_data.and_then(|(_, _, p, _, _, _, _)| {
+            description: profile_data.and_then(|(_, _, _, p, _, _, _, _)| {
                 p.as_ref().and_then(|prof| prof.description.clone())
             }),
             // TODO: Profile timestamps should be fetched from records_literal table via record_id FK

parakeet/src/xrpc/app_bsky/unspecced/mod.rs

···
                     // Check if has profile (display name or description)
                     profiles
                         .get(did)
-                        .and_then(|p| p.2.as_ref())
+                        .and_then(|p| p.3.as_ref()) // .3 is Option<models::Profile>
                         .map(|prof| prof.display_name.is_some() || prof.description.is_some())
                         .unwrap_or(false)
                 })