Reseed lost event history from the KV projection on transition

A record whose event log is empty (the EVENTS stream was purged, or
the record predates event-sourcing) but whose projection exists used
to be stranded - replay found nothing, every transition failed with
"unknown item". transition_or_reseed reconstructs a minimal one-event
history from the projection's own current state and retries, making
"zap the whole event stream" a recoverable operation by design instead
of a slow-motion data loss. Reseed races are guarded by the same
expected-sequence-0 CAS create uses.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/aggregates/mod.rs

@@ -240,6 +240,39 @@ pub async fn create(
     })
 }
 
+/// Reconstructs a minimal one-event history for an aggregate whose
+/// event log is empty but whose KV projection still knows its current
+/// state - what makes wiping the EVENTS stream a recoverable
+/// operation rather than one that strands every pre-wipe record in an
+/// untransitionable limbo. `expected_seq: Some(0)` guards two racing
+/// reseeds the same way `create` guards a double submit.
+pub async fn reseed(
+    js: &async_nats::jetstream::Context,
+    schema: &AggregateSchema,
+    id: &str,
+    state: &str,
+    occurred_at_ms: i64,
+) -> Result<(), TransitionError> {
+    let event_type = schema.event_for_state.get(state).ok_or_else(|| {
+        TransitionError::Store(format!(
+            "cannot reseed {id}: state {state:?} is not declared for bucket {:?}",
+            schema.bucket
+        ))
+    })?;
+    append_event(
+        js,
+        &schema.bucket,
+        id,
+        event_type,
+        serde_json::json!({ "reseeded_from_projection": true }),
+        occurred_at_ms,
+        Some(0),
+    )
+    .await
+    .map_err(classify_publish_err)?;
+    Ok(())
+}
+
 /// Loads and replays `id`'s current state, then - if `target` is
 /// actually reachable from it - appends the corresponding event with
 /// the aggregate's just-replayed sequence as the CAS guard.

src/answers.rs

@@ -154,6 +154,39 @@ pub async fn transition_answers(
     }
 }
 
+/// `aggregates::transition`, with one recovery path: an aggregate the
+/// event log has never heard of but whose KV projection exists (the
+/// log was purged, or predates event-sourcing) gets a minimal history
+/// reseeded from the projection's own current state, then the
+/// transition retries. Keeps "zap the whole event stream" a safe
+/// operation instead of one that strands every pre-wipe record.
+#[cfg(feature = "ssr")]
+async fn transition_or_reseed(
+    js: &async_nats::jetstream::Context,
+    schema: &crate::aggregates::AggregateSchema,
+    item_id: &str,
+    to: &str,
+    current_state: &str,
+    payload: serde_json::Value,
+    occurred_at_ms: i64,
+) -> Result<(), crate::aggregates::TransitionError> {
+    match crate::aggregates::transition(js, schema, item_id, to, payload.clone(), occurred_at_ms)
+        .await
+    {
+        Err(crate::aggregates::TransitionError::UnknownAggregate) => {
+            tracing::warn!(
+                bucket = %schema.bucket, item_id, current_state,
+                "no event history for a projected record - reseeding from the projection"
+            );
+            crate::aggregates::reseed(js, schema, item_id, current_state, occurred_at_ms).await?;
+            crate::aggregates::transition(js, schema, item_id, to, payload, occurred_at_ms)
+                .await
+                .map(|_| ())
+        }
+        r => r.map(|_| ()),
+    }
+}
+
 /// Same alternative-scoped feature lookup as `get_resource`, and for
 /// the same reason - a feature name isn't unique across a whole
 /// question, only within its own alternative.
@@ -230,9 +263,17 @@ async fn apply_transition(
     // the direct KV mutate this always did - not every resource has to
     // be event-sourced to keep working.
     if let Some(schema) = state.aggregates.load().get(bucket) {
-        crate::aggregates::transition(&state.jetstream, schema, &item.item_id, &item.to, decision_payload_for_event.clone(), decided_ms)
+        transition_or_reseed(
-            .await
+            &state.jetstream,
-            .map_err(|e| ServerFnError::new(e.to_string()))?;
+            schema,
+            &item.item_id,
+            &item.to,
+            &answer.state,
+            decision_payload_for_event.clone(),
+            decided_ms,
+        )
+        .await
+        .map_err(|e| ServerFnError::new(e.to_string()))?;
     }
 
     answer.state = item.to.clone();
@@ -327,11 +368,12 @@ pub async fn self_transition_answer(
         serde_json::json!({ "to": self_transition.to, "item": item_id, "by": email });
 
     if let Some(schema) = state.aggregates.load().get(&self_transition.bucket) {
-        crate::aggregates::transition(
+        transition_or_reseed(
             &state.jetstream,
             schema,
             &item_id,
             &self_transition.to,
+            &answer.state,
             decision_payload_for_event.clone(),
             decided_ms,
         )