Attempts to resolve a Webhook source from a provided `database.schema.name` path. Returns a struct that can be used to append data to the underlying storate collection, and the types we should cast the request to.
(
coord: &mut Coordinator,
database: String,
schema: String,
name: String,
)
| 1983 | /// Returns a struct that can be used to append data to the underlying storate collection, and the |
| 1984 | /// types we should cast the request to. |
| 1985 | fn resolve( |
| 1986 | coord: &mut Coordinator, |
| 1987 | database: String, |
| 1988 | schema: String, |
| 1989 | name: String, |
| 1990 | ) -> Result<AppendWebhookResponse, PartialItemName> { |
| 1991 | // Resolve our collection. |
| 1992 | let name = PartialItemName { |
| 1993 | database: Some(database), |
| 1994 | schema: Some(schema), |
| 1995 | item: name, |
| 1996 | }; |
| 1997 | let Ok(entry) = coord |
| 1998 | .catalog() |
| 1999 | .resolve_entry(None, &vec![], &name, &SYSTEM_CONN_ID) |
| 2000 | else { |
| 2001 | return Err(name); |
| 2002 | }; |
| 2003 | |
| 2004 | // Webhooks can be created with `CREATE SOURCE` or `CREATE TABLE`. |
| 2005 | let (data_source, desc, global_id) = match entry.item() { |
| 2006 | CatalogItem::Source(Source { |
| 2007 | data_source: data_source @ DataSourceDesc::Webhook { .. }, |
| 2008 | desc, |
| 2009 | global_id, |
| 2010 | .. |
| 2011 | }) => (data_source, desc.clone(), *global_id), |
| 2012 | CatalogItem::Table( |
| 2013 | table @ Table { |
| 2014 | desc, |
| 2015 | data_source: |
| 2016 | TableDataSource::DataSource { |
| 2017 | desc: data_source @ DataSourceDesc::Webhook { .. }, |
| 2018 | .. |
| 2019 | }, |
| 2020 | .. |
| 2021 | }, |
| 2022 | ) => (data_source, desc.latest(), table.global_id_writes()), |
| 2023 | _ => return Err(name), |
| 2024 | }; |
| 2025 | |
| 2026 | let DataSourceDesc::Webhook { |
| 2027 | validate_using, |
| 2028 | body_format, |
| 2029 | headers, |
| 2030 | .. |
| 2031 | } = data_source |
| 2032 | else { |
| 2033 | mz_ore::soft_panic_or_log!("programming error! checked above for webhook"); |
| 2034 | return Err(name); |
| 2035 | }; |
| 2036 | let body_format = body_format.clone(); |
| 2037 | let header_tys = headers.clone(); |
| 2038 | |
| 2039 | // Assert we have one column for the body, and how ever many are required for |
| 2040 | // the headers. |
| 2041 | let num_columns = headers.num_columns() + 1; |
| 2042 | mz_ore::soft_assert_or_log!( |