canton_pqs/
row.rs

1//! What PQS returns, typed.
2//!
3//! The payload is stored in the Daml JSON encoding — which is what
4//! `canton-daml` implements — so it deserializes into the same generated type
5//! a transaction stream yields. The columns around it are the ledger metadata
6//! PQS records: who signed, when it was created, at which offset, and when it
7//! was archived if it has been.
8
9use canton_core::{Error, Result};
10use canton_daml::ContractId;
11
12/// A contract as PQS holds it: the typed payload, plus the ledger facts about
13/// it.
14// `contract_id` reads as a repetition of the type name, but it is what the
15// ledger calls this and what the PQS column is called. Shortening it to `id`
16// would make the accessor say less than the domain does.
17#[allow(clippy::struct_field_names)]
18#[derive(Clone, Debug)]
19pub struct Contract<T> {
20    contract_id: ContractId<T>,
21    payload: T,
22    created_at_offset: i64,
23    archived_at_offset: Option<i64>,
24    created_effective_at: Option<time::OffsetDateTime>,
25    archived_effective_at: Option<time::OffsetDateTime>,
26    signatories: Vec<String>,
27    observers: Vec<String>,
28    witnesses: Vec<String>,
29    package_name: String,
30    package_version: String,
31    package_id: String,
32}
33
34impl<T> Contract<T> {
35    /// The contract id, typed.
36    #[must_use]
37    pub fn contract_id(&self) -> &ContractId<T> {
38        &self.contract_id
39    }
40
41    /// The payload.
42    #[must_use]
43    pub fn payload(&self) -> &T {
44        &self.payload
45    }
46
47    /// Take the payload.
48    #[must_use]
49    pub fn into_payload(self) -> T {
50        self.payload
51    }
52
53    /// The offset the contract was created at.
54    #[must_use]
55    pub fn created_at_offset(&self) -> i64 {
56        self.created_at_offset
57    }
58
59    /// The offset it was archived at, if it has been archived at all.
60    ///
61    /// **Being set does not mean the contract was inactive for the read that
62    /// returned it.** `active_at(o)` selects contracts whose lifetime *covers*
63    /// `o`, so a contract that was live then and archived afterwards comes back
64    /// with this set — filtering those out would drop exactly the rows the
65    /// query asked for. Compare it against the offset that was read at:
66    /// [`was_active_at`](Self::was_active_at) does that.
67    #[must_use]
68    pub fn archived_at_offset(&self) -> Option<i64> {
69        self.archived_at_offset
70    }
71
72    /// Whether the contract was active at `offset`.
73    ///
74    /// There is no offset-free answer to "is this active": it depends on when
75    /// you are asking about. A contract read by `active_at(o)` was active at
76    /// `o` whatever this says about later offsets.
77    #[must_use]
78    pub fn was_active_at(&self, offset: i64) -> bool {
79        self.created_at_offset <= offset
80            && self
81                .archived_at_offset
82                .is_none_or(|archived| archived > offset)
83    }
84
85    /// Ledger effective time of the creation.
86    #[must_use]
87    pub fn created_effective_at(&self) -> Option<time::OffsetDateTime> {
88        self.created_effective_at
89    }
90
91    /// Ledger effective time of the archival.
92    #[must_use]
93    pub fn archived_effective_at(&self) -> Option<time::OffsetDateTime> {
94        self.archived_effective_at
95    }
96
97    /// The parties that signed it.
98    #[must_use]
99    pub fn signatories(&self) -> &[String] {
100        &self.signatories
101    }
102
103    /// The parties that observe it.
104    #[must_use]
105    pub fn observers(&self) -> &[String] {
106        &self.observers
107    }
108
109    /// Every party that witnessed it — the column `Predicate::witness` filters
110    /// on, so selecting by it and then not being able to see it made no sense.
111    #[must_use]
112    pub fn witnesses(&self) -> &[String] {
113        &self.witnesses
114    }
115
116    /// The Daml package name — stable across an upgrade.
117    #[must_use]
118    pub fn package_name(&self) -> &str {
119        &self.package_name
120    }
121
122    /// The package version.
123    #[must_use]
124    pub fn package_version(&self) -> &str {
125        &self.package_version
126    }
127
128    /// The package id — the exact build, which a version does not pin.
129    #[must_use]
130    pub fn package_id(&self) -> &str {
131        &self.package_id
132    }
133}
134
135/// An exercised choice as PQS holds it.
136#[derive(Clone, Debug)]
137pub struct Exercise<C> {
138    contract_id: String,
139    choice: String,
140    consuming: bool,
141    argument: C,
142    result: serde_json::Value,
143    exercised_at_offset: i64,
144    exercised_effective_at: Option<time::OffsetDateTime>,
145    controllers: Vec<String>,
146}
147
148impl<C> Exercise<C> {
149    /// The contract the choice was exercised on.
150    #[must_use]
151    pub fn contract_id(&self) -> &str {
152        &self.contract_id
153    }
154
155    /// The choice name.
156    #[must_use]
157    pub fn choice(&self) -> &str {
158        &self.choice
159    }
160
161    /// Whether exercising it archived the contract.
162    #[must_use]
163    pub fn consuming(&self) -> bool {
164        self.consuming
165    }
166
167    /// The typed argument.
168    #[must_use]
169    pub fn argument(&self) -> &C {
170        &self.argument
171    }
172
173    /// The result, untyped.
174    ///
175    /// A choice's return type is not the argument's, and PQS does not say which
176    /// it is — so this stays JSON rather than pretending to a type it cannot
177    /// know. Decode it with [`result_as`](Self::result_as).
178    #[must_use]
179    pub fn result(&self) -> &serde_json::Value {
180        &self.result
181    }
182
183    /// Decode the result into a type the caller knows it to be.
184    ///
185    /// # Errors
186    /// [`Error::Payload`] if it does not match.
187    pub fn result_as<R: serde::de::DeserializeOwned>(&self) -> Result<R> {
188        serde_json::from_value(self.result.clone()).map_err(|e| Error::Payload(Box::new(e)))
189    }
190
191    /// The offset it was exercised at.
192    #[must_use]
193    pub fn exercised_at_offset(&self) -> i64 {
194        self.exercised_at_offset
195    }
196
197    /// Ledger effective time of the exercise.
198    #[must_use]
199    pub fn exercised_effective_at(&self) -> Option<time::OffsetDateTime> {
200        self.exercised_effective_at
201    }
202
203    /// The parties that exercised it.
204    #[must_use]
205    pub fn controllers(&self) -> &[String] {
206        &self.controllers
207    }
208}
209
210/// Read a column by name, saying which one was missing rather than panicking.
211///
212/// PQS's row shape is a contract of its own; a rename upstream should say so
213/// by name, not as an index that no longer exists.
214fn column<'a, T: tokio_postgres::types::FromSql<'a>>(
215    row: &'a tokio_postgres::Row,
216    name: &str,
217) -> Result<T> {
218    row.try_get(name).map_err(|e| {
219        Error::UnexpectedResponse(format!("PQS returned no usable `{name}` column: {e}"))
220    })
221}
222
223impl<T: serde::de::DeserializeOwned> Contract<T> {
224    /// Decode one row.
225    ///
226    /// # Errors
227    /// [`Error::UnexpectedResponse`] if a column is missing or of another type,
228    /// [`Error::Payload`] if the payload does not match `T`.
229    pub fn from_row(row: &tokio_postgres::Row) -> Result<Self> {
230        let payload: serde_json::Value = column(row, "payload")?;
231        let contract_id: String = column(row, "contract_id")?;
232        Ok(Self {
233            payload: serde_json::from_value(payload).map_err(|e| {
234                Error::Payload(Box::new(std::io::Error::other(format!(
235                    "contract {contract_id}: {e}"
236                ))))
237            })?,
238            contract_id: ContractId::new(contract_id),
239            created_at_offset: column(row, "created_at_offset")?,
240            archived_at_offset: column(row, "archived_at_offset")?,
241            created_effective_at: column(row, "created_effective_at")?,
242            archived_effective_at: column(row, "archived_effective_at")?,
243            signatories: column(row, "signatories")?,
244            observers: column(row, "observers")?,
245            witnesses: column(row, "witnesses")?,
246            package_name: column(row, "package_name")?,
247            package_version: column(row, "package_version")?,
248            package_id: column(row, "package_id")?,
249        })
250    }
251}
252
253impl<C: serde::de::DeserializeOwned> Exercise<C> {
254    /// Decode one row.
255    ///
256    /// # Errors
257    /// As [`Contract::from_row`].
258    pub fn from_row(row: &tokio_postgres::Row) -> Result<Self> {
259        let argument: serde_json::Value = column(row, "argument")?;
260        let choice: String = column(row, "choice")?;
261        Ok(Self {
262            argument: serde_json::from_value(argument).map_err(|e| {
263                Error::Payload(Box::new(std::io::Error::other(format!(
264                    "choice {choice}: {e}"
265                ))))
266            })?,
267            choice,
268            contract_id: column(row, "contract_id")?,
269            consuming: column(row, "consuming")?,
270            result: column::<Option<serde_json::Value>>(row, "result")?
271                .unwrap_or(serde_json::Value::Null),
272            exercised_at_offset: column(row, "exercised_at_offset")?,
273            exercised_effective_at: column(row, "exercised_effective_at")?,
274            controllers: column(row, "controllers")?,
275        })
276    }
277}
278
279#[cfg(test)]
280#[allow(clippy::unwrap_used, clippy::expect_used)]
281mod tests {
282    use super::*;
283
284    fn contract(created: i64, archived: Option<i64>) -> Contract<()> {
285        Contract {
286            contract_id: ContractId::new("00c"),
287            payload: (),
288            created_at_offset: created,
289            archived_at_offset: archived,
290            created_effective_at: Some(time::OffsetDateTime::UNIX_EPOCH),
291            archived_effective_at: archived.map(|_| time::OffsetDateTime::UNIX_EPOCH),
292            signatories: vec!["alice".to_string()],
293            observers: vec!["bob".to_string()],
294            witnesses: vec!["carol".to_string()],
295            package_name: "quickstart-licensing".to_string(),
296            package_version: "0.1.0".to_string(),
297            package_id: "abc123".to_string(),
298        }
299    }
300
301    /// Active at `o` means created at or before `o` and not archived at or
302    /// before it. Every boundary, both sides.
303    #[test]
304    fn was_active_at_is_inclusive_of_creation_and_exclusive_of_archival() {
305        let live = contract(10, None);
306        assert!(!live.was_active_at(9));
307        assert!(live.was_active_at(10));
308        assert!(live.was_active_at(1_000));
309
310        let archived = contract(10, Some(20));
311        assert!(!archived.was_active_at(9));
312        assert!(archived.was_active_at(10));
313        assert!(archived.was_active_at(19));
314        assert!(
315            !archived.was_active_at(20),
316            "archived *at* 20 is not active at 20"
317        );
318        assert!(!archived.was_active_at(21));
319    }
320
321    /// The accessors return what the row carried, not a default.
322    #[test]
323    fn the_accessors_return_the_row() {
324        let c = contract(10, Some(20));
325        assert_eq!(c.contract_id().as_str(), "00c");
326        assert_eq!(c.created_at_offset(), 10);
327        assert_eq!(c.archived_at_offset(), Some(20));
328        assert!(c.created_effective_at().is_some());
329        assert!(c.archived_effective_at().is_some());
330        assert_eq!(c.signatories(), ["alice".to_string()]);
331        assert_eq!(c.observers(), ["bob".to_string()]);
332        assert_eq!(c.witnesses(), ["carol".to_string()]);
333        assert_eq!(c.package_name(), "quickstart-licensing");
334        assert_eq!(c.package_version(), "0.1.0");
335        assert_eq!(c.package_id(), "abc123");
336        assert_eq!(*c.payload(), ());
337        assert_eq!(contract(1, None).archived_effective_at(), None);
338
339        let e = Exercise {
340            contract_id: "00x".to_string(),
341            choice: "Accept".to_string(),
342            consuming: true,
343            argument: 7_i64,
344            result: serde_json::json!({ "ok": true }),
345            exercised_at_offset: 33,
346            exercised_effective_at: Some(time::OffsetDateTime::UNIX_EPOCH),
347            controllers: vec!["alice".to_string()],
348        };
349        assert_eq!(e.contract_id(), "00x");
350        assert_eq!(e.choice(), "Accept");
351        assert!(e.consuming());
352        assert_eq!(*e.argument(), 7);
353        assert_eq!(e.result(), &serde_json::json!({ "ok": true }));
354        assert_eq!(e.exercised_at_offset(), 33);
355        assert!(e.exercised_effective_at().is_some());
356        assert_eq!(e.controllers(), ["alice".to_string()]);
357
358        let non_consuming = Exercise {
359            consuming: false,
360            exercised_effective_at: None,
361            ..e
362        };
363        assert!(!non_consuming.consuming());
364        assert_eq!(non_consuming.exercised_effective_at(), None);
365    }
366}
canton_pqs/row.rs

canton_pqs/
row.rs
