1use super::routing::*;
2use super::*;
3
4pub(crate) fn filter_inbound_circle_changeset(
5 conn: &Connection,
6 changeset: &[u8],
7 circle_id: CircleId,
8 store_transitions: &StoreAudienceTransitions,
9 gates: &Gates,
10 routing_key: &RowRoutingKey,
11) -> Result<Vec<u8>, GateError> {
12 unsafe {
13 let package_audience = Audience::Circle(circle_id);
14 let (normalized, _) = normalize_inbound_private_routes_raw(
15 conn,
16 changeset,
17 &package_audience,
18 store_transitions,
19 gates,
20 routing_key,
21 )?;
22 filter_inbound_audience_rows_raw(conn, &normalized, &package_audience, gates, routing_key)
23 }
24}
25
26pub(crate) fn filter_inbound_store_rows(
27 conn: &Connection,
28 changeset: &[u8],
29 gates: &Gates,
30 routing_key: &RowRoutingKey,
31) -> Result<Vec<u8>, GateError> {
32 unsafe {
33 filter_inbound_audience_rows_raw(conn, changeset, &Audience::Store, gates, routing_key)
34 }
35}
36
37pub(crate) fn align_inbound_scoped_root_audiences(
38 conn: &Connection,
39 changeset: &[u8],
40 package_audience: &Audience,
41 gates: &Gates,
42 routing_key: &RowRoutingKey,
43) -> Result<(), GateError> {
44 unsafe {
45 for_each_change(changeset, |_iter, row| {
46 if row.op == ffi::SQLITE_DELETE {
47 return Ok(());
48 }
49 let Some(TableGate::ScopedRoot { audience_col }) = gates.tables.get(&row.table) else {
50 return Ok(());
51 };
52 let row_id = row
53 .pk()
54 .ok_or_else(|| GateError::MissingChangesetPrimaryKey(row.table.clone()))?;
55 let routing_id = row_routing_id(routing_key, &row.table, row_id).to_string();
56 let winning_audience = winning_store_audience(conn, &routing_id)?;
57 if winning_audience.as_ref() != Some(package_audience) {
58 return Err(GateError::InvalidInboundAudiencePackage(format!(
59 "eligible {}.{row_id} package no longer matches its winning Store audience",
60 row.table
61 )));
62 }
63 let updated = conn
64 .execute(
65 &format!(
66 "UPDATE {} SET {} = ?1 WHERE id = ?2",
67 quote_ident(&row.table),
68 quote_ident(&audience_col.name),
69 ),
70 rusqlite::params![package_audience.column_value(), row_id],
71 )
72 .map_err(|source| {
73 GateError::Sql(
74 format!("align inbound audience for {}.{row_id}", row.table),
75 source,
76 )
77 })?;
78 if updated != 1 {
79 return Err(GateError::InvalidInboundAudiencePackage(format!(
80 "eligible {}.{row_id} did not materialize exactly one scoped root",
81 row.table
82 )));
83 }
84 Ok(())
85 })
86 }
87}
88
89pub(crate) fn winning_store_audience(
90 conn: &Connection,
91 routing_id: &str,
92) -> Result<Option<Audience>, GateError> {
93 query_row_optional(
94 conn,
95 "SELECT circle_id FROM _coven_audience WHERE routing_id = ?1",
96 [routing_id],
97 |record| record.get::<_, Option<String>>(0),
98 )?
99 .map(|circle_id| {
100 Audience::from_column(circle_id.as_deref()).map_err(|source| {
101 GateError::InvalidInboundAudienceEncoding {
102 context: format!("winning Store audience for {routing_id} is invalid"),
103 source,
104 }
105 })
106 })
107 .transpose()
108}
109
110pub(crate) fn normalize_inbound_store_changeset(
111 conn: &Connection,
112 changeset: &[u8],
113 gates: &Gates,
114 routing_key: &RowRoutingKey,
115) -> Result<InboundStoreChangesets, GateError> {
116 let store_transitions = store_audience_transitions(changeset)?;
117 let normalized = unsafe {
118 normalize_inbound_private_routes_raw(
119 conn,
120 changeset,
121 &Audience::Store,
122 &store_transitions,
123 gates,
124 routing_key,
125 )
126 .map(|(normalized, _)| normalized)?
127 };
128 unsafe {
129 let mirror = Changegroup::new()?;
130 mirror.set_schema(conn.handle())?;
131 let rows = Changegroup::new()?;
132 rows.set_schema(conn.handle())?;
133 for_each_change(&normalized, |iter, row| {
134 if row.table == "_coven_audience" {
135 mirror.add_change(iter)
136 } else {
137 rows.add_change(iter)
138 }
139 })?;
140 Ok(InboundStoreChangesets {
141 mirror: mirror.output()?,
142 rows: rows.output()?,
143 })
144 }
145}
146
147pub(crate) type PackageRoutes = HashMap<(String, String), (String, String)>;
148
149pub fn store_audience_transitions(changeset: &[u8]) -> Result<StoreAudienceTransitions, GateError> {
150 let mut transitions = StoreAudienceTransitions::default();
151 unsafe {
152 for_each_change(changeset, |_iter, row| {
153 if row.table != "_coven_audience"
154 || (row.op != ffi::SQLITE_INSERT && row.op != ffi::SQLITE_UPDATE)
155 {
156 return Ok(());
157 }
158 let routing_id = row
159 .pk()
160 .ok_or_else(|| GateError::MissingChangesetPrimaryKey(row.table.clone()))?;
161 let circle_id = row.new_value(1).ok_or_else(|| {
162 GateError::InvalidInboundAudiencePackage(format!(
163 "Store audience transition {routing_id} has no audience"
164 ))
165 })?;
166 let audience = Audience::from_column(circle_id).map_err(|source| {
167 GateError::InvalidInboundAudienceEncoding {
168 context: format!(
169 "Store audience transition {routing_id} has an invalid audience"
170 ),
171 source,
172 }
173 })?;
174 if audience == Audience::Local {
175 return Err(GateError::InvalidInboundAudiencePackage(format!(
176 "Store audience transition {routing_id} has a Local audience"
177 )));
178 }
179 let stamp = row.new_value(2).flatten().ok_or_else(|| {
180 GateError::InvalidInboundAudiencePackage(format!(
181 "Store audience transition {routing_id} has no _updated_at"
182 ))
183 })?;
184 if transitions
185 .by_routing_id
186 .insert(routing_id.to_string(), (audience, stamp.to_string()))
187 .is_some()
188 {
189 return Err(GateError::InvalidInboundAudiencePackage(format!(
190 "Store package contains duplicate audience transitions for {routing_id}"
191 )));
192 }
193 Ok(())
194 })?;
195 }
196 Ok(transitions)
197}
198
199pub(crate) unsafe fn filter_inbound_audience_rows_raw(
200 conn: &Connection,
201 changeset: &[u8],
202 package_audience: &Audience,
203 gates: &Gates,
204 routing_key: &RowRoutingKey,
205) -> Result<Vec<u8>, GateError> {
206 let allow_unscoped = package_audience == &Audience::Store;
207 for_each_change(changeset, |_iter, row| {
208 if row.table == "_coven_audience" {
209 return Err(GateError::InvalidInboundAudiencePackage(
210 "audience row package contains the Store audience mirror".to_string(),
211 ));
212 }
213 if row.table != "_coven_row_routes" && !gates.table_is_scoped(&row.table) && !allow_unscoped
214 {
215 return Err(GateError::InvalidInboundAudiencePackage(format!(
216 "Circle package contains unscoped table {}",
217 row.table
218 )));
219 }
220 if row.op != ffi::SQLITE_DELETE {
221 if let Some(TableGate::ScopedRoot { audience_col }) = gates.tables.get(&row.table) {
222 if let Some(value) = row.new_value(audience_col.index) {
223 let row_audience = Audience::from_column(value).map_err(|source| {
224 GateError::InvalidInboundAudienceEncoding {
225 context: format!("scoped row {} has an invalid audience", row.table),
226 source,
227 }
228 })?;
229 if &row_audience != package_audience {
230 return Err(GateError::InvalidInboundAudiencePackage(format!(
231 "scoped row {} is packaged for a different audience than its row value",
232 row.table
233 )));
234 }
235 }
236 }
237 }
238 Ok(())
239 })?;
240
241 let group = Changegroup::new()?;
242 group.set_schema(conn.handle())?;
243 for_each_change(changeset, |iter, row| {
244 if row.table != "_coven_row_routes" && !gates.table_is_scoped(&row.table) {
245 group.add_change(iter)?;
246 return Ok(());
247 }
248 let routing_id = if row.table == "_coven_row_routes" {
249 row.pk()
250 .ok_or_else(|| GateError::MissingChangesetPrimaryKey(row.table.clone()))?
251 .to_string()
252 } else {
253 let row_id = row
254 .pk()
255 .ok_or_else(|| GateError::MissingChangesetPrimaryKey(row.table.clone()))?;
256 row_routing_id(routing_key, &row.table, row_id).to_string()
257 };
258 let winning_audience = winning_store_audience(conn, &routing_id)?;
259 if winning_audience.as_ref() == Some(package_audience) {
260 group.add_change(iter)?;
261 }
262 Ok(())
263 })?;
264 group.output()
265}
266
267pub(crate) unsafe fn normalize_inbound_private_routes_raw(
268 conn: &Connection,
269 changeset: &[u8],
270 package_audience: &Audience,
271 store_transitions: &StoreAudienceTransitions,
272 gates: &Gates,
273 routing_key: &RowRoutingKey,
274) -> Result<(Vec<u8>, PackageRoutes), GateError> {
275 let package_routes = validate_inbound_private_routes_raw(
276 conn,
277 changeset,
278 package_audience,
279 store_transitions,
280 gates,
281 routing_key,
282 )?;
283 let group = Changegroup::new()?;
284 group.set_schema(conn.handle())?;
285 for_each_change(changeset, |iter, row| {
286 if row.table != "_coven_row_routes" {
287 group.add_change(iter)?;
288 }
289 Ok(())
290 })?;
291 let mut rows = package_routes
292 .iter()
293 .map(|((table, row_id), (routing_id, stamp))| {
294 (
295 routing_id.clone(),
296 table.clone(),
297 row_id.clone(),
298 stamp.clone(),
299 )
300 })
301 .collect::<Vec<_>>();
302 rows.sort();
303 let canonical_routes = private_route_insert_changeset(&rows)?;
304 for_each_change(&canonical_routes, |iter, _row| group.add_change(iter))?;
305 Ok((group.output()?, package_routes))
306}
307
308pub(crate) unsafe fn validate_inbound_private_routes_raw(
309 conn: &Connection,
310 changeset: &[u8],
311 package_audience: &Audience,
312 store_transitions: &StoreAudienceTransitions,
313 gates: &Gates,
314 routing_key: &RowRoutingKey,
315) -> Result<PackageRoutes, GateError> {
316 let mut package_routes = PackageRoutes::new();
317 let mut package_row_inserts = HashSet::<(String, String)>::new();
318 for_each_change(changeset, |_iter, row| {
319 if row.table == "_coven_audience" {
320 return Ok(());
321 }
322 if row.table != "_coven_row_routes" {
323 if gates.table_is_scoped(&row.table) && row.op == ffi::SQLITE_INSERT {
324 let row_id = row
325 .pk()
326 .ok_or_else(|| GateError::MissingChangesetPrimaryKey(row.table.clone()))?;
327 let columns = crate::gate::gate_table_columns(conn, &row.table)?;
328 let stamp_index = columns
329 .iter()
330 .position(|column| column == "_updated_at")
331 .ok_or_else(|| {
332 GateError::MissingFkColumn(row.table.clone(), "_updated_at".to_string())
333 })?;
334 row.new_value(stamp_index).flatten().ok_or_else(|| {
335 GateError::InvalidInboundAudiencePackage(format!(
336 "complete row INSERT {}.{row_id} has no _updated_at",
337 row.table
338 ))
339 })?;
340 package_row_inserts.insert((row.table.clone(), row_id.to_string()));
341 }
342 return Ok(());
343 }
344 if row.op != ffi::SQLITE_INSERT {
345 return Err(GateError::InvalidInboundAudiencePackage(
346 "private routes must be complete INSERT images".to_string(),
347 ));
348 }
349 let routing_id = row.new_value(0).flatten().ok_or_else(|| {
350 GateError::InvalidInboundAudiencePackage(
351 "private route INSERT has no routing id".to_string(),
352 )
353 })?;
354 let table = row.new_value(1).flatten().ok_or_else(|| {
355 GateError::InvalidInboundAudiencePackage("private route has no table name".to_string())
356 })?;
357 let row_id = row.new_value(2).flatten().ok_or_else(|| {
358 GateError::InvalidInboundAudiencePackage("private route has no row id".to_string())
359 })?;
360 let stamp = row.new_value(3).flatten().ok_or_else(|| {
361 GateError::InvalidInboundAudiencePackage("private route has no _updated_at".to_string())
362 })?;
363 if !gates.table_is_scoped(table) {
364 return Err(GateError::InvalidInboundAudiencePackage(format!(
365 "private route names unscoped table {table}"
366 )));
367 }
368 let identity = gates.row_identity(table).ok_or_else(|| {
369 GateError::InvalidInboundAudiencePackage(format!(
370 "private route names undeclared table {table}"
371 ))
372 })?;
373 identity.validate(table, row_id).map_err(|source| {
374 GateError::InvalidInboundRowIdentity {
375 context: "private route row identity is invalid".to_string(),
376 source,
377 }
378 })?;
379 let expected_routing_id = row_routing_id(routing_key, table, row_id).to_string();
380 if routing_id != expected_routing_id {
381 return Err(GateError::InvalidInboundAudiencePackage(format!(
382 "private route id does not authenticate {table}.{row_id}"
383 )));
384 }
385 if package_routes
386 .insert(
387 (table.to_string(), row_id.to_string()),
388 (routing_id.to_string(), stamp.to_string()),
389 )
390 .is_some()
391 {
392 return Err(GateError::InvalidInboundAudiencePackage(format!(
393 "duplicate private route for {table}.{row_id}"
394 )));
395 }
396 Ok(())
397 })?;
398 for (row, (routing_id, route_stamp)) in &package_routes {
399 if !package_row_inserts.contains(row) {
400 return Err(GateError::InvalidInboundAudiencePackage(format!(
401 "private route for {}.{} has no complete row INSERT",
402 row.0, row.1
403 )));
404 }
405 let (transition_audience, audience_stamp) = store_transitions
406 .by_routing_id
407 .get(routing_id)
408 .ok_or_else(|| {
409 GateError::InvalidInboundAudiencePackage(format!(
410 "private route for {}.{} has no Store audience transition",
411 row.0, row.1
412 ))
413 })?;
414 if transition_audience != package_audience {
415 return Err(GateError::InvalidInboundAudiencePackage(format!(
416 "private route for {}.{} is packaged for a different audience than its Store transition",
417 row.0, row.1
418 )));
419 }
420 if route_stamp != audience_stamp {
421 return Err(GateError::InvalidInboundAudiencePackage(format!(
422 "private route for {}.{} has a different _updated_at than its Store audience transition",
423 row.0, row.1
424 )));
425 }
426 }
427 for row in &package_row_inserts {
428 if package_routes.contains_key(row) {
429 continue;
430 }
431 let existing = query_row_optional(
432 conn,
433 "SELECT routing_id FROM _coven_row_routes
434 WHERE table_name = ?1 AND row_id = ?2",
435 (&row.0, &row.1),
436 |record| record.get::<_, String>(0),
437 )?
438 .ok_or_else(|| {
439 GateError::InvalidInboundAudiencePackage(format!(
440 "scoped row INSERT {}.{} has no private route",
441 row.0, row.1
442 ))
443 })?;
444 let expected = row_routing_id(routing_key, &row.0, &row.1).to_string();
445 if existing != expected {
446 return Err(GateError::InvalidInboundAudiencePackage(format!(
447 "stored private route does not authenticate {}.{}",
448 row.0, row.1
449 )));
450 }
451 }
452 Ok(package_routes)
453}