| 1 | import * as Effect from "effect/Effect"; |
| 2 | import * as SqlClient from "effect/sql/SqlClient"; |
| 3 | |
| 4 | // Server auto-settlement used to stamp settledAt with the sweep time instead |
| 5 | // of the thread's last activity. Repair the projection only: the engine and |
| 6 | // projectors bootstrap from projection rows and cursors, never a full replay, |
| 7 | // so the historical event payloads can stay as they were recorded. The |
| 8 | // decider stamped settledAt and occurred_at from the same clock read, which |
| 9 | // is how an unrepaired automatic settlement is identified below. |
| 10 | export default Effect.gen(function* () { |
| 11 | const sql = yield* SqlClient.SqlClient; |
| 12 | |
| 13 | yield* sql` |
| 14 | WITH activity_timestamps AS ( |
| 15 | SELECT thread_id, created_at AS activity_at |
| 16 | FROM projection_thread_messages |
| 17 | WHERE role = 'user' |
| 18 | UNION ALL |
| 19 | SELECT thread_id, requested_at |
| 20 | FROM projection_turns |
| 21 | UNION ALL |
| 22 | SELECT thread_id, started_at |
| 23 | FROM projection_turns |
| 24 | WHERE started_at IS NOT NULL |
| 25 | UNION ALL |
| 26 | SELECT thread_id, completed_at |
| 27 | FROM projection_turns |
| 28 | WHERE completed_at IS NOT NULL |
| 29 | ), |
| 30 | automatic_settlements AS ( |
| 31 | SELECT |
| 32 | stream_id AS thread_id, |
| 33 | occurred_at, |
| 34 | json_extract(payload_json, '$.settledAt') AS settled_at |
| 35 | FROM orchestration_events |
| 36 | WHERE aggregate_kind = 'thread' |
| 37 | AND event_type = 'thread.settled' |
| 38 | AND actor_kind = 'server' |
| 39 | AND command_id LIKE 'server:auto-settle:%' |
| 40 | AND json_type(payload_json, '$.settledAt') = 'text' |
| 41 | AND json_extract(payload_json, '$.settledAt') = occurred_at |
| 42 | ) |
| 43 | UPDATE projection_threads AS thread |
| 44 | SET settled_at = ( |
| 45 | SELECT COALESCE( |
| 46 | ( |
| 47 | SELECT activity.activity_at |
| 48 | FROM activity_timestamps AS activity |
| 49 | WHERE activity.thread_id = thread.thread_id |
| 50 | AND julianday(activity.activity_at) IS NOT NULL |
| 51 | AND julianday(activity.activity_at) <= julianday(automatic.occurred_at) |
| 52 | ORDER BY julianday(activity.activity_at) DESC |
| 53 | LIMIT 1 |
| 54 | ), |
| 55 | thread.created_at |
| 56 | ) |
| 57 | FROM automatic_settlements AS automatic |
| 58 | WHERE automatic.thread_id = thread.thread_id |
| 59 | AND automatic.settled_at = thread.settled_at |
| 60 | LIMIT 1 |
| 61 | ) |
| 62 | WHERE thread.settled_override = 'settled' |
| 63 | AND EXISTS ( |
| 64 | SELECT 1 |
| 65 | FROM automatic_settlements AS automatic |
| 66 | WHERE automatic.thread_id = thread.thread_id |
| 67 | AND automatic.settled_at = thread.settled_at |
| 68 | ) |
| 69 | `; |
| 70 | }); |