Skip to main content

chio_store_sqlite/receipt_store/
liability_claims.rs

1use super::*;
2
3/// Liability claim workflow artifacts embed the full signed chain of prior
4/// artifacts (claim -> response -> dispute -> adjudication ->
5/// payout_instruction -> payout_receipt -> settlement_instruction ->
6/// settlement_receipt), each carrying the one before it. Deserializing and
7/// structurally comparing these deeply nested values (to validate that a new
8/// artifact references the persisted prior stage unchanged) can require more
9/// stack than the single writer actor's thread allots by default. Running
10/// that work on a scoped worker thread with a larger stack keeps the writer
11/// actor's own thread untouched while staying safe as the chain grows.
12fn run_with_generous_stack<F, T>(job: F) -> Result<T, ReceiptStoreError>
13where
14    F: FnOnce() -> Result<T, ReceiptStoreError> + Send,
15    T: Send,
16{
17    std::thread::scope(|scope| {
18        let handle = std::thread::Builder::new()
19            .stack_size(16 * 1024 * 1024)
20            .spawn_scoped(scope, job)?;
21        handle.join().map_err(|_| {
22            ReceiptStoreError::Canonical(
23                "liability claim workflow worker thread panicked".to_string(),
24            )
25        })?
26    })
27}
28
29impl SqliteReceiptStore {
30    pub fn record_liability_claim_package(
31        &mut self,
32        claim: &SignedLiabilityClaimPackage,
33    ) -> Result<(), ReceiptStoreError> {
34        if !claim
35            .verify_signature()
36            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
37        {
38            return Err(ReceiptStoreError::Conflict(
39                "liability claim package signature verification failed".to_string(),
40            ));
41        }
42        claim.body.validate().map_err(ReceiptStoreError::Conflict)?;
43
44        let claim_owned = claim.clone();
45        self.writer_handle().run_write(move |connection| {
46        run_with_generous_stack(move || {
47        let claim = &claim_owned;
48        let artifact = &claim.body;
49        let tx = connection.transaction()?;
50        let existing = tx
51            .query_row(
52                "SELECT claim_id FROM liability_claim_packages WHERE claim_id = ?1",
53                params![artifact.claim_id],
54                |row| row.get::<_, String>(0),
55            )
56            .optional()?;
57        if existing.is_some() {
58            return Err(ReceiptStoreError::Conflict(format!(
59                "liability claim package `{}` already exists",
60                artifact.claim_id
61            )));
62        }
63
64        let stored_bound_raw_json = tx
65            .query_row(
66                "SELECT raw_json
67                 FROM liability_bound_coverages
68                 WHERE bound_coverage_id = ?1",
69                params![artifact.bound_coverage.body.bound_coverage_id],
70                |row| row.get::<_, String>(0),
71            )
72            .optional()?
73            .ok_or_else(|| {
74                ReceiptStoreError::NotFound(format!(
75                    "liability bound coverage `{}` not found",
76                    artifact.bound_coverage.body.bound_coverage_id
77                ))
78            })?;
79        let stored_bound: SignedLiabilityBoundCoverage =
80            serde_json::from_str(&stored_bound_raw_json)?;
81        if stored_bound.body != artifact.bound_coverage.body {
82            return Err(ReceiptStoreError::Conflict(
83                "liability claim package bound_coverage does not match the persisted bound coverage"
84                    .to_string(),
85            ));
86        }
87
88        let stored_bond_raw_json = tx
89            .query_row(
90                "SELECT raw_json
91                 FROM credit_bonds
92                 WHERE bond_id = ?1",
93                params![artifact.bond.body.bond_id],
94                |row| row.get::<_, String>(0),
95            )
96            .optional()?
97            .ok_or_else(|| {
98                ReceiptStoreError::NotFound(format!(
99                    "credit bond `{}` not found",
100                    artifact.bond.body.bond_id
101                ))
102            })?;
103        let stored_bond: SignedCreditBond = serde_json::from_str(&stored_bond_raw_json)?;
104        if stored_bond.body != artifact.bond.body {
105            return Err(ReceiptStoreError::Conflict(
106                "liability claim package bond does not match the persisted credit bond".to_string(),
107            ));
108        }
109
110        let stored_loss_raw_json = tx
111            .query_row(
112                "SELECT raw_json
113                 FROM credit_loss_lifecycle
114                 WHERE event_id = ?1",
115                params![artifact.loss_event.body.event_id],
116                |row| row.get::<_, String>(0),
117            )
118            .optional()?
119            .ok_or_else(|| {
120                ReceiptStoreError::NotFound(format!(
121                    "credit loss lifecycle event `{}` not found",
122                    artifact.loss_event.body.event_id
123                ))
124            })?;
125        let stored_loss: SignedCreditLossLifecycle = serde_json::from_str(&stored_loss_raw_json)?;
126        if stored_loss.body != artifact.loss_event.body {
127            return Err(ReceiptStoreError::Conflict(
128                "liability claim package loss_event does not match the persisted credit loss lifecycle event"
129                    .to_string(),
130            ));
131        }
132
133        for receipt_id in &artifact.receipt_ids {
134            let exists = tx
135                .query_row(
136                    "SELECT 1 FROM chio_tool_receipts WHERE receipt_id = ?1",
137                    params![receipt_id],
138                    |row| row.get::<_, i64>(0),
139                )
140                .optional()?;
141            if exists.is_none() {
142                return Err(ReceiptStoreError::NotFound(format!(
143                    "receipt {receipt_id} does not exist"
144                )));
145            }
146        }
147
148        tx.execute(
149            "INSERT INTO liability_claim_packages (
150                claim_id, issued_at, provider_id, policy_number, jurisdiction, subject_key,
151                claim_event_at, raw_json, signer_key, signature
152             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
153            params![
154                artifact.claim_id,
155                artifact.issued_at as i64,
156                artifact
157                    .bound_coverage
158                    .body
159                    .placement
160                    .body
161                    .quote_response
162                    .body
163                    .quote_request
164                    .body
165                    .provider_policy
166                    .provider_id,
167                artifact.bound_coverage.body.policy_number,
168                artifact
169                    .bound_coverage
170                    .body
171                    .placement
172                    .body
173                    .quote_response
174                    .body
175                    .quote_request
176                    .body
177                    .provider_policy
178                    .jurisdiction,
179                artifact
180                    .bound_coverage
181                    .body
182                    .placement
183                    .body
184                    .quote_response
185                    .body
186                    .quote_request
187                    .body
188                    .risk_package
189                    .body
190                    .subject_key,
191                artifact.claim_event_at as i64,
192                serde_json::to_string(claim)?,
193                claim.signer_key.to_hex(),
194                claim.signature.to_hex(),
195            ],
196        )?;
197
198        tx.commit()?;
199        Ok(())
200        })
201        })
202    }
203
204    pub fn record_liability_claim_response(
205        &mut self,
206        response: &SignedLiabilityClaimResponse,
207    ) -> Result<(), ReceiptStoreError> {
208        if !response
209            .verify_signature()
210            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
211        {
212            return Err(ReceiptStoreError::Conflict(
213                "liability claim response signature verification failed".to_string(),
214            ));
215        }
216        response
217            .body
218            .validate()
219            .map_err(ReceiptStoreError::Conflict)?;
220
221        let response_owned = response.clone();
222        self.writer_handle().run_write(move |connection| {
223            run_with_generous_stack(move || {
224                let response = &response_owned;
225                let artifact = &response.body;
226                let tx = connection.transaction()?;
227                let existing = tx
228                    .query_row(
229                        "SELECT claim_response_id
230                 FROM liability_claim_responses
231                 WHERE claim_response_id = ?1",
232                        params![artifact.claim_response_id],
233                        |row| row.get::<_, String>(0),
234                    )
235                    .optional()?;
236                if existing.is_some() {
237                    return Err(ReceiptStoreError::Conflict(format!(
238                        "liability claim response `{}` already exists",
239                        artifact.claim_response_id
240                    )));
241                }
242
243                let stored_claim_raw_json = tx
244                    .query_row(
245                        "SELECT raw_json
246                 FROM liability_claim_packages
247                 WHERE claim_id = ?1",
248                        params![artifact.claim.body.claim_id],
249                        |row| row.get::<_, String>(0),
250                    )
251                    .optional()?
252                    .ok_or_else(|| {
253                        ReceiptStoreError::NotFound(format!(
254                            "liability claim package `{}` not found",
255                            artifact.claim.body.claim_id
256                        ))
257                    })?;
258                let stored_claim: SignedLiabilityClaimPackage =
259                    serde_json::from_str(&stored_claim_raw_json)?;
260                if stored_claim.body != artifact.claim.body {
261                    return Err(ReceiptStoreError::Conflict(
262                        "liability claim response claim does not match the persisted claim package"
263                            .to_string(),
264                    ));
265                }
266
267                tx.execute(
268                    "INSERT INTO liability_claim_responses (
269                claim_response_id, issued_at, claim_id, provider_id, disposition,
270                raw_json, signer_key, signature
271             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
272                    params![
273                        artifact.claim_response_id,
274                        artifact.issued_at as i64,
275                        artifact.claim.body.claim_id,
276                        artifact
277                            .claim
278                            .body
279                            .bound_coverage
280                            .body
281                            .placement
282                            .body
283                            .quote_response
284                            .body
285                            .quote_request
286                            .body
287                            .provider_policy
288                            .provider_id,
289                        serde_json::to_string(&artifact.disposition)?,
290                        serde_json::to_string(response)?,
291                        response.signer_key.to_hex(),
292                        response.signature.to_hex(),
293                    ],
294                )?;
295
296                tx.commit()?;
297                Ok(())
298            })
299        })
300    }
301
302    pub fn record_liability_claim_dispute(
303        &mut self,
304        dispute: &SignedLiabilityClaimDispute,
305    ) -> Result<(), ReceiptStoreError> {
306        if !dispute
307            .verify_signature()
308            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
309        {
310            return Err(ReceiptStoreError::Conflict(
311                "liability claim dispute signature verification failed".to_string(),
312            ));
313        }
314        dispute
315            .body
316            .validate()
317            .map_err(ReceiptStoreError::Conflict)?;
318
319        let dispute_owned = dispute.clone();
320        self.writer_handle().run_write(move |connection| {
321        run_with_generous_stack(move || {
322        let dispute = &dispute_owned;
323        let artifact = &dispute.body;
324        let tx = connection.transaction()?;
325        let existing = tx
326            .query_row(
327                "SELECT dispute_id FROM liability_claim_disputes WHERE dispute_id = ?1",
328                params![artifact.dispute_id],
329                |row| row.get::<_, String>(0),
330            )
331            .optional()?;
332        if existing.is_some() {
333            return Err(ReceiptStoreError::Conflict(format!(
334                "liability claim dispute `{}` already exists",
335                artifact.dispute_id
336            )));
337        }
338
339        let stored_response_raw_json = tx
340            .query_row(
341                "SELECT raw_json
342                 FROM liability_claim_responses
343                 WHERE claim_response_id = ?1",
344                params![artifact.provider_response.body.claim_response_id],
345                |row| row.get::<_, String>(0),
346            )
347            .optional()?
348            .ok_or_else(|| {
349                ReceiptStoreError::NotFound(format!(
350                    "liability claim response `{}` not found",
351                    artifact.provider_response.body.claim_response_id
352                ))
353            })?;
354        let stored_response: SignedLiabilityClaimResponse =
355            serde_json::from_str(&stored_response_raw_json)?;
356        if stored_response.body != artifact.provider_response.body {
357            return Err(ReceiptStoreError::Conflict(
358                "liability claim dispute provider_response does not match the persisted claim response"
359                    .to_string(),
360            ));
361        }
362
363        tx.execute(
364            "INSERT INTO liability_claim_disputes (
365                dispute_id, issued_at, claim_id, claim_response_id, provider_id,
366                raw_json, signer_key, signature
367             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
368            params![
369                artifact.dispute_id,
370                artifact.issued_at as i64,
371                artifact.provider_response.body.claim.body.claim_id,
372                artifact.provider_response.body.claim_response_id,
373                artifact
374                    .provider_response
375                    .body
376                    .claim
377                    .body
378                    .bound_coverage
379                    .body
380                    .placement
381                    .body
382                    .quote_response
383                    .body
384                    .quote_request
385                    .body
386                    .provider_policy
387                    .provider_id,
388                serde_json::to_string(dispute)?,
389                dispute.signer_key.to_hex(),
390                dispute.signature.to_hex(),
391            ],
392        )?;
393
394        tx.commit()?;
395        Ok(())
396        })
397        })
398    }
399
400    pub fn record_liability_claim_adjudication(
401        &mut self,
402        adjudication: &SignedLiabilityClaimAdjudication,
403    ) -> Result<(), ReceiptStoreError> {
404        if !adjudication
405            .verify_signature()
406            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
407        {
408            return Err(ReceiptStoreError::Conflict(
409                "liability claim adjudication signature verification failed".to_string(),
410            ));
411        }
412        adjudication
413            .body
414            .validate()
415            .map_err(ReceiptStoreError::Conflict)?;
416
417        let adjudication_owned = adjudication.clone();
418        self.writer_handle().run_write(move |connection| {
419            run_with_generous_stack(move || {
420                let adjudication = &adjudication_owned;
421                let artifact = &adjudication.body;
422                let tx = connection.transaction()?;
423                let existing = tx
424                    .query_row(
425                        "SELECT adjudication_id
426                 FROM liability_claim_adjudications
427                 WHERE adjudication_id = ?1",
428                        params![artifact.adjudication_id],
429                        |row| row.get::<_, String>(0),
430                    )
431                    .optional()?;
432                if existing.is_some() {
433                    return Err(ReceiptStoreError::Conflict(format!(
434                        "liability claim adjudication `{}` already exists",
435                        artifact.adjudication_id
436                    )));
437                }
438
439                let stored_dispute_raw_json = tx
440                    .query_row(
441                        "SELECT raw_json
442                 FROM liability_claim_disputes
443                 WHERE dispute_id = ?1",
444                        params![artifact.dispute.body.dispute_id],
445                        |row| row.get::<_, String>(0),
446                    )
447                    .optional()?
448                    .ok_or_else(|| {
449                        ReceiptStoreError::NotFound(format!(
450                            "liability claim dispute `{}` not found",
451                            artifact.dispute.body.dispute_id
452                        ))
453                    })?;
454                let stored_dispute: SignedLiabilityClaimDispute =
455                    serde_json::from_str(&stored_dispute_raw_json)?;
456                if stored_dispute.body != artifact.dispute.body {
457                    return Err(ReceiptStoreError::Conflict(
458                "liability claim adjudication dispute does not match the persisted claim dispute"
459                    .to_string(),
460            ));
461                }
462
463                tx.execute(
464                    "INSERT INTO liability_claim_adjudications (
465                adjudication_id, issued_at, claim_id, dispute_id, outcome,
466                raw_json, signer_key, signature
467             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
468                    params![
469                        artifact.adjudication_id,
470                        artifact.issued_at as i64,
471                        artifact
472                            .dispute
473                            .body
474                            .provider_response
475                            .body
476                            .claim
477                            .body
478                            .claim_id,
479                        artifact.dispute.body.dispute_id,
480                        serde_json::to_string(&artifact.outcome)?,
481                        serde_json::to_string(adjudication)?,
482                        adjudication.signer_key.to_hex(),
483                        adjudication.signature.to_hex(),
484                    ],
485                )?;
486
487                tx.commit()?;
488                Ok(())
489            })
490        })
491    }
492
493    pub fn record_liability_claim_payout_instruction(
494        &mut self,
495        payout_instruction: &SignedLiabilityClaimPayoutInstruction,
496    ) -> Result<(), ReceiptStoreError> {
497        if !payout_instruction
498            .verify_signature()
499            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
500        {
501            return Err(ReceiptStoreError::Conflict(
502                "liability claim payout instruction signature verification failed".to_string(),
503            ));
504        }
505        payout_instruction
506            .body
507            .validate()
508            .map_err(ReceiptStoreError::Conflict)?;
509
510        let payout_instruction_owned = payout_instruction.clone();
511        self.writer_handle().run_write(move |connection| {
512        run_with_generous_stack(move || {
513        let payout_instruction = &payout_instruction_owned;
514        let artifact = &payout_instruction.body;
515        let tx = connection.transaction()?;
516        let existing = tx
517            .query_row(
518                "SELECT payout_instruction_id
519                 FROM liability_claim_payout_instructions
520                 WHERE payout_instruction_id = ?1",
521                params![artifact.payout_instruction_id],
522                |row| row.get::<_, String>(0),
523            )
524            .optional()?;
525        if existing.is_some() {
526            return Err(ReceiptStoreError::Conflict(format!(
527                "liability claim payout instruction `{}` already exists",
528                artifact.payout_instruction_id
529            )));
530        }
531
532        let stored_adjudication_raw_json = tx
533            .query_row(
534                "SELECT raw_json
535                 FROM liability_claim_adjudications
536                 WHERE adjudication_id = ?1",
537                params![artifact.adjudication.body.adjudication_id],
538                |row| row.get::<_, String>(0),
539            )
540            .optional()?
541            .ok_or_else(|| {
542                ReceiptStoreError::NotFound(format!(
543                    "liability claim adjudication `{}` not found",
544                    artifact.adjudication.body.adjudication_id
545                ))
546            })?;
547        let stored_adjudication: SignedLiabilityClaimAdjudication =
548            serde_json::from_str(&stored_adjudication_raw_json)?;
549        if stored_adjudication.body != artifact.adjudication.body {
550            return Err(ReceiptStoreError::Conflict(
551                "liability claim payout instruction adjudication does not match the persisted adjudication"
552                    .to_string(),
553            ));
554        }
555
556        let claim_id = artifact
557            .adjudication
558            .body
559            .dispute
560            .body
561            .provider_response
562            .body
563            .claim
564            .body
565            .claim_id
566            .clone();
567
568        tx.execute(
569            "INSERT INTO liability_claim_payout_instructions (
570                payout_instruction_id, issued_at, claim_id, adjudication_id,
571                payout_amount_units, payout_amount_currency,
572                raw_json, signer_key, signature
573             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
574            params![
575                artifact.payout_instruction_id,
576                artifact.issued_at as i64,
577                claim_id,
578                artifact.adjudication.body.adjudication_id,
579                artifact.payout_amount.units as i64,
580                artifact.payout_amount.currency,
581                serde_json::to_string(payout_instruction)?,
582                payout_instruction.signer_key.to_hex(),
583                payout_instruction.signature.to_hex(),
584            ],
585        )?;
586
587        tx.commit()?;
588        Ok(())
589        })
590        })
591    }
592
593    pub fn record_liability_claim_payout_receipt(
594        &mut self,
595        payout_receipt: &SignedLiabilityClaimPayoutReceipt,
596    ) -> Result<(), ReceiptStoreError> {
597        if !payout_receipt
598            .verify_signature()
599            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
600        {
601            return Err(ReceiptStoreError::Conflict(
602                "liability claim payout receipt signature verification failed".to_string(),
603            ));
604        }
605        payout_receipt
606            .body
607            .validate()
608            .map_err(ReceiptStoreError::Conflict)?;
609
610        let payout_receipt_owned = payout_receipt.clone();
611        self.writer_handle().run_write(move |connection| {
612        run_with_generous_stack(move || {
613        let payout_receipt = &payout_receipt_owned;
614        let artifact = &payout_receipt.body;
615        let tx = connection.transaction()?;
616        let existing = tx
617            .query_row(
618                "SELECT payout_receipt_id
619                 FROM liability_claim_payout_receipts
620                 WHERE payout_receipt_id = ?1",
621                params![artifact.payout_receipt_id],
622                |row| row.get::<_, String>(0),
623            )
624            .optional()?;
625        if existing.is_some() {
626            return Err(ReceiptStoreError::Conflict(format!(
627                "liability claim payout receipt `{}` already exists",
628                artifact.payout_receipt_id
629            )));
630        }
631
632        let stored_instruction_raw_json = tx
633            .query_row(
634                "SELECT raw_json
635                 FROM liability_claim_payout_instructions
636                 WHERE payout_instruction_id = ?1",
637                params![artifact.payout_instruction.body.payout_instruction_id],
638                |row| row.get::<_, String>(0),
639            )
640            .optional()?
641            .ok_or_else(|| {
642                ReceiptStoreError::NotFound(format!(
643                    "liability claim payout instruction `{}` not found",
644                    artifact.payout_instruction.body.payout_instruction_id
645                ))
646            })?;
647        let stored_instruction: SignedLiabilityClaimPayoutInstruction =
648            serde_json::from_str(&stored_instruction_raw_json)?;
649        if stored_instruction.body != artifact.payout_instruction.body {
650            return Err(ReceiptStoreError::Conflict(
651                "liability claim payout receipt payout_instruction does not match the persisted payout instruction"
652                    .to_string(),
653            ));
654        }
655
656        let claim_id = artifact
657            .payout_instruction
658            .body
659            .adjudication
660            .body
661            .dispute
662            .body
663            .provider_response
664            .body
665            .claim
666            .body
667            .claim_id
668            .clone();
669
670        tx.execute(
671            "INSERT INTO liability_claim_payout_receipts (
672                payout_receipt_id, issued_at, claim_id, payout_instruction_id,
673                reconciliation_state, raw_json, signer_key, signature
674             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
675            params![
676                artifact.payout_receipt_id,
677                artifact.issued_at as i64,
678                claim_id,
679                artifact.payout_instruction.body.payout_instruction_id,
680                serde_json::to_string(&artifact.reconciliation_state)?,
681                serde_json::to_string(payout_receipt)?,
682                payout_receipt.signer_key.to_hex(),
683                payout_receipt.signature.to_hex(),
684            ],
685        )?;
686
687        tx.commit()?;
688        Ok(())
689        })
690        })
691    }
692
693    pub fn record_liability_claim_settlement_instruction(
694        &mut self,
695        settlement_instruction: &SignedLiabilityClaimSettlementInstruction,
696    ) -> Result<(), ReceiptStoreError> {
697        if !settlement_instruction
698            .verify_signature()
699            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
700        {
701            return Err(ReceiptStoreError::Conflict(
702                "liability claim settlement instruction signature verification failed".to_string(),
703            ));
704        }
705        settlement_instruction
706            .body
707            .validate()
708            .map_err(ReceiptStoreError::Conflict)?;
709
710        let settlement_instruction_owned = settlement_instruction.clone();
711        self.writer_handle().run_write(move |connection| {
712        run_with_generous_stack(move || {
713        let settlement_instruction = &settlement_instruction_owned;
714        let artifact = &settlement_instruction.body;
715        let tx = connection.transaction()?;
716        let existing = tx
717            .query_row(
718                "SELECT settlement_instruction_id
719                 FROM liability_claim_settlement_instructions
720                 WHERE settlement_instruction_id = ?1",
721                params![artifact.settlement_instruction_id],
722                |row| row.get::<_, String>(0),
723            )
724            .optional()?;
725        if existing.is_some() {
726            return Err(ReceiptStoreError::Conflict(format!(
727                "liability claim settlement instruction `{}` already exists",
728                artifact.settlement_instruction_id
729            )));
730        }
731
732        let stored_payout_receipt_raw_json = tx
733            .query_row(
734                "SELECT raw_json
735                 FROM liability_claim_payout_receipts
736                 WHERE payout_receipt_id = ?1",
737                params![artifact.payout_receipt.body.payout_receipt_id],
738                |row| row.get::<_, String>(0),
739            )
740            .optional()?
741            .ok_or_else(|| {
742                ReceiptStoreError::NotFound(format!(
743                    "liability claim payout receipt `{}` not found",
744                    artifact.payout_receipt.body.payout_receipt_id
745                ))
746            })?;
747        let stored_payout_receipt: SignedLiabilityClaimPayoutReceipt =
748            serde_json::from_str(&stored_payout_receipt_raw_json)?;
749        if stored_payout_receipt.body != artifact.payout_receipt.body {
750            return Err(ReceiptStoreError::Conflict(
751                "liability claim settlement instruction payout_receipt does not match the persisted payout receipt"
752                    .to_string(),
753            ));
754        }
755
756        let claim_id = artifact
757            .payout_receipt
758            .body
759            .payout_instruction
760            .body
761            .adjudication
762            .body
763            .dispute
764            .body
765            .provider_response
766            .body
767            .claim
768            .body
769            .claim_id
770            .clone();
771
772        tx.execute(
773            "INSERT INTO liability_claim_settlement_instructions (
774                settlement_instruction_id, issued_at, claim_id, payout_receipt_id,
775                settlement_kind, payer_role, payer_id, payee_role, payee_id,
776                settlement_amount_units, settlement_amount_currency,
777                raw_json, signer_key, signature
778             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)",
779            params![
780                artifact.settlement_instruction_id,
781                artifact.issued_at as i64,
782                claim_id,
783                artifact.payout_receipt.body.payout_receipt_id,
784                serde_json::to_string(&artifact.settlement_kind)?,
785                serde_json::to_string(&artifact.topology.payer.role)?,
786                artifact.topology.payer.party_id,
787                serde_json::to_string(&artifact.topology.payee.role)?,
788                artifact.topology.payee.party_id,
789                artifact.settlement_amount.units as i64,
790                artifact.settlement_amount.currency,
791                serde_json::to_string(settlement_instruction)?,
792                settlement_instruction.signer_key.to_hex(),
793                settlement_instruction.signature.to_hex(),
794            ],
795        )?;
796
797        tx.commit()?;
798        Ok(())
799        })
800        })
801    }
802
803    pub fn record_liability_claim_settlement_receipt(
804        &mut self,
805        settlement_receipt: &SignedLiabilityClaimSettlementReceipt,
806    ) -> Result<(), ReceiptStoreError> {
807        if !settlement_receipt
808            .verify_signature()
809            .map_err(|error| ReceiptStoreError::Canonical(error.to_string()))?
810        {
811            return Err(ReceiptStoreError::Conflict(
812                "liability claim settlement receipt signature verification failed".to_string(),
813            ));
814        }
815        settlement_receipt
816            .body
817            .validate()
818            .map_err(ReceiptStoreError::Conflict)?;
819
820        let settlement_receipt_owned = settlement_receipt.clone();
821        self.writer_handle().run_write(move |connection| {
822        run_with_generous_stack(move || {
823        let settlement_receipt = &settlement_receipt_owned;
824        let artifact = &settlement_receipt.body;
825        let tx = connection.transaction()?;
826        let existing = tx
827            .query_row(
828                "SELECT settlement_receipt_id
829                 FROM liability_claim_settlement_receipts
830                 WHERE settlement_receipt_id = ?1",
831                params![artifact.settlement_receipt_id],
832                |row| row.get::<_, String>(0),
833            )
834            .optional()?;
835        if existing.is_some() {
836            return Err(ReceiptStoreError::Conflict(format!(
837                "liability claim settlement receipt `{}` already exists",
838                artifact.settlement_receipt_id
839            )));
840        }
841
842        let stored_instruction_raw_json = tx
843            .query_row(
844                "SELECT raw_json
845                 FROM liability_claim_settlement_instructions
846                 WHERE settlement_instruction_id = ?1",
847                params![
848                    artifact
849                        .settlement_instruction
850                        .body
851                        .settlement_instruction_id
852                ],
853                |row| row.get::<_, String>(0),
854            )
855            .optional()?
856            .ok_or_else(|| {
857                ReceiptStoreError::NotFound(format!(
858                    "liability claim settlement instruction `{}` not found",
859                    artifact
860                        .settlement_instruction
861                        .body
862                        .settlement_instruction_id
863                ))
864            })?;
865        let stored_instruction: SignedLiabilityClaimSettlementInstruction =
866            serde_json::from_str(&stored_instruction_raw_json)?;
867        if stored_instruction.body != artifact.settlement_instruction.body {
868            return Err(ReceiptStoreError::Conflict(
869                "liability claim settlement receipt settlement_instruction does not match the persisted settlement instruction"
870                    .to_string(),
871            ));
872        }
873
874        let claim_id = artifact
875            .settlement_instruction
876            .body
877            .payout_receipt
878            .body
879            .payout_instruction
880            .body
881            .adjudication
882            .body
883            .dispute
884            .body
885            .provider_response
886            .body
887            .claim
888            .body
889            .claim_id
890            .clone();
891
892        tx.execute(
893            "INSERT INTO liability_claim_settlement_receipts (
894                settlement_receipt_id, issued_at, claim_id, settlement_instruction_id,
895                reconciliation_state, raw_json, signer_key, signature
896             ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
897            params![
898                artifact.settlement_receipt_id,
899                artifact.issued_at as i64,
900                claim_id,
901                artifact
902                    .settlement_instruction
903                    .body
904                    .settlement_instruction_id,
905                serde_json::to_string(&artifact.reconciliation_state)?,
906                serde_json::to_string(settlement_receipt)?,
907                settlement_receipt.signer_key.to_hex(),
908                settlement_receipt.signature.to_hex(),
909            ],
910        )?;
911
912        tx.commit()?;
913        Ok(())
914        })
915        })
916    }
917
918    pub fn query_liability_claim_workflows(
919        &self,
920        query: &LiabilityClaimWorkflowQuery,
921    ) -> Result<LiabilityClaimWorkflowReport, ReceiptStoreError> {
922        let normalized = query.normalized();
923        let connection = self.connection()?;
924        let mut statement = connection.prepare(
925            "SELECT raw_json
926             FROM liability_claim_packages
927             ORDER BY issued_at DESC, claim_id DESC",
928        )?;
929        let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
930
931        let mut matching_claims = 0_u64;
932        let mut provider_responses = 0_u64;
933        let mut accepted_responses = 0_u64;
934        let mut denied_responses = 0_u64;
935        let mut disputes = 0_u64;
936        let mut adjudications = 0_u64;
937        let mut payout_instructions = 0_u64;
938        let mut payout_receipts = 0_u64;
939        let mut matched_payout_receipts = 0_u64;
940        let mut mismatched_payout_receipts = 0_u64;
941        let mut settlement_instructions = 0_u64;
942        let mut settlement_receipts = 0_u64;
943        let mut matched_settlement_receipts = 0_u64;
944        let mut mismatched_settlement_receipts = 0_u64;
945        let mut counterparty_mismatch_settlement_receipts = 0_u64;
946        let mut claims = Vec::new();
947
948        for row in rows {
949            let raw_json = row?;
950            let claim: SignedLiabilityClaimPackage = serde_json::from_str(&raw_json)?;
951            if !liability_claim_workflow_matches_query(&claim, &normalized) {
952                continue;
953            }
954            matching_claims += 1;
955
956            let provider_response = self
957                .connection()?
958                .query_row(
959                    "SELECT raw_json
960                     FROM liability_claim_responses
961                     WHERE claim_id = ?1
962                     ORDER BY issued_at DESC, claim_response_id DESC
963                     LIMIT 1",
964                    params![claim.body.claim_id],
965                    |row| row.get::<_, String>(0),
966                )
967                .optional()?
968                .map(|raw_json| serde_json::from_str::<SignedLiabilityClaimResponse>(&raw_json))
969                .transpose()?;
970            if let Some(response) = provider_response.as_ref() {
971                provider_responses += 1;
972                match response.body.disposition {
973                    LiabilityClaimResponseDisposition::Accepted => accepted_responses += 1,
974                    LiabilityClaimResponseDisposition::Denied => denied_responses += 1,
975                    LiabilityClaimResponseDisposition::Acknowledged => {}
976                }
977            }
978
979            let dispute = self
980                .connection()?
981                .query_row(
982                    "SELECT raw_json
983                     FROM liability_claim_disputes
984                     WHERE claim_id = ?1
985                     ORDER BY issued_at DESC, dispute_id DESC
986                     LIMIT 1",
987                    params![claim.body.claim_id],
988                    |row| row.get::<_, String>(0),
989                )
990                .optional()?
991                .map(|raw_json| serde_json::from_str::<SignedLiabilityClaimDispute>(&raw_json))
992                .transpose()?;
993            if dispute.is_some() {
994                disputes += 1;
995            }
996
997            let adjudication = self
998                .connection()?
999                .query_row(
1000                    "SELECT raw_json
1001                     FROM liability_claim_adjudications
1002                     WHERE claim_id = ?1
1003                     ORDER BY issued_at DESC, adjudication_id DESC
1004                     LIMIT 1",
1005                    params![claim.body.claim_id],
1006                    |row| row.get::<_, String>(0),
1007                )
1008                .optional()?
1009                .map(|raw_json| serde_json::from_str::<SignedLiabilityClaimAdjudication>(&raw_json))
1010                .transpose()?;
1011            if adjudication.is_some() {
1012                adjudications += 1;
1013            }
1014
1015            let payout_instruction = self
1016                .connection()?
1017                .query_row(
1018                    "SELECT raw_json
1019                     FROM liability_claim_payout_instructions
1020                     WHERE claim_id = ?1
1021                     ORDER BY issued_at DESC, payout_instruction_id DESC
1022                     LIMIT 1",
1023                    params![claim.body.claim_id],
1024                    |row| row.get::<_, String>(0),
1025                )
1026                .optional()?
1027                .map(|raw_json| {
1028                    serde_json::from_str::<SignedLiabilityClaimPayoutInstruction>(&raw_json)
1029                })
1030                .transpose()?;
1031            if payout_instruction.is_some() {
1032                payout_instructions += 1;
1033            }
1034
1035            let payout_receipt = self
1036                .connection()?
1037                .query_row(
1038                    "SELECT raw_json
1039                     FROM liability_claim_payout_receipts
1040                     WHERE claim_id = ?1
1041                     ORDER BY issued_at DESC, payout_receipt_id DESC
1042                     LIMIT 1",
1043                    params![claim.body.claim_id],
1044                    |row| row.get::<_, String>(0),
1045                )
1046                .optional()?
1047                .map(|raw_json| {
1048                    serde_json::from_str::<SignedLiabilityClaimPayoutReceipt>(&raw_json)
1049                })
1050                .transpose()?;
1051            if let Some(receipt) = payout_receipt.as_ref() {
1052                payout_receipts += 1;
1053                match receipt.body.reconciliation_state {
1054                    LiabilityClaimPayoutReconciliationState::Matched => {
1055                        matched_payout_receipts += 1;
1056                    }
1057                    LiabilityClaimPayoutReconciliationState::AmountMismatch => {
1058                        mismatched_payout_receipts += 1;
1059                    }
1060                }
1061            }
1062
1063            let settlement_instruction = self
1064                .connection()?
1065                .query_row(
1066                    "SELECT raw_json
1067                     FROM liability_claim_settlement_instructions
1068                     WHERE claim_id = ?1
1069                     ORDER BY issued_at DESC, settlement_instruction_id DESC
1070                     LIMIT 1",
1071                    params![claim.body.claim_id],
1072                    |row| row.get::<_, String>(0),
1073                )
1074                .optional()?
1075                .map(|raw_json| {
1076                    serde_json::from_str::<SignedLiabilityClaimSettlementInstruction>(&raw_json)
1077                })
1078                .transpose()?;
1079            if settlement_instruction.is_some() {
1080                settlement_instructions += 1;
1081            }
1082
1083            let settlement_receipt = self
1084                .connection()?
1085                .query_row(
1086                    "SELECT raw_json
1087                     FROM liability_claim_settlement_receipts
1088                     WHERE claim_id = ?1
1089                     ORDER BY issued_at DESC, settlement_receipt_id DESC
1090                     LIMIT 1",
1091                    params![claim.body.claim_id],
1092                    |row| row.get::<_, String>(0),
1093                )
1094                .optional()?
1095                .map(|raw_json| {
1096                    serde_json::from_str::<SignedLiabilityClaimSettlementReceipt>(&raw_json)
1097                })
1098                .transpose()?;
1099            if let Some(receipt) = settlement_receipt.as_ref() {
1100                settlement_receipts += 1;
1101                match receipt.body.reconciliation_state {
1102                    LiabilityClaimSettlementReconciliationState::Matched => {
1103                        matched_settlement_receipts += 1;
1104                    }
1105                    LiabilityClaimSettlementReconciliationState::AmountMismatch => {
1106                        mismatched_settlement_receipts += 1;
1107                    }
1108                    LiabilityClaimSettlementReconciliationState::CounterpartyMismatch => {
1109                        counterparty_mismatch_settlement_receipts += 1;
1110                    }
1111                }
1112            }
1113
1114            if claims.len() < normalized.limit_or_default() {
1115                claims.push(LiabilityClaimWorkflowRow {
1116                    claim,
1117                    provider_response,
1118                    dispute,
1119                    adjudication,
1120                    payout_instruction,
1121                    payout_receipt,
1122                    settlement_instruction,
1123                    settlement_receipt,
1124                });
1125            }
1126        }
1127
1128        Ok(LiabilityClaimWorkflowReport {
1129            schema: LIABILITY_CLAIM_WORKFLOW_REPORT_SCHEMA.to_string(),
1130            generated_at: unix_now(),
1131            query: normalized,
1132            summary: LiabilityClaimWorkflowSummary {
1133                matching_claims,
1134                returned_claims: claims.len() as u64,
1135                provider_responses,
1136                accepted_responses,
1137                denied_responses,
1138                disputes,
1139                adjudications,
1140                payout_instructions,
1141                payout_receipts,
1142                matched_payout_receipts,
1143                mismatched_payout_receipts,
1144                settlement_instructions,
1145                settlement_receipts,
1146                matched_settlement_receipts,
1147                mismatched_settlement_receipts,
1148                counterparty_mismatch_settlement_receipts,
1149            },
1150            claims,
1151        })
1152    }
1153}