use crate::config::{Config, ConfigError};
use crate::events::{AnyError, EventBus, EventHandler, EventName};
use crate::ports::{Port, Ports};
use crate::scheduled::ScheduledBudget;
use crate::surface::Surface;
use crate::template::TemplateRegistry;
use crate::venture::Venture;
use std::sync::Arc;
pub use futures_core::future::BoxFuture;
pub const HARNESS_API: u32 = 1;
#[must_use]
pub fn harness_api_mismatch(module: &dyn Module) -> String {
format!(
"module `{name}` v{version} targets harness API {api}, but {core} v{core_version} \
provides harness API {harness_api}: rebuild `{name}` against this core — the supported \
ranges are in docs/COMPATIBILITY.md",
name = module.name(),
version = module.version(),
api = module.harness_api(),
core = env!("CARGO_PKG_NAME"),
core_version = env!("CARGO_PKG_VERSION"),
harness_api = HARNESS_API,
)
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct SqlMigration {
pub id: &'static str,
pub name: &'static str,
pub sql: &'static str,
pub transactional: bool,
}
impl SqlMigration {
#[must_use]
pub const fn new(id: &'static str, name: &'static str, sql: &'static str) -> Self {
Self {
id,
name,
sql,
transactional: true,
}
}
#[must_use]
pub const fn non_transactional(mut self) -> Self {
self.transactional = false;
self
}
}
pub const fn assert_migration_set(set: &[SqlMigration]) {
let mut index = 0;
while index < set.len() {
let id = set[index].id.as_bytes();
assert!(
id.len() == 4,
"a migration id is four digits, zero-padded: `0001`, not `1`"
);
let mut digit = 0;
while digit < 4 {
assert!(
id[digit] >= b'0' && id[digit] <= b'9',
"a migration id is four digits, zero-padded: `0001`, not `init`"
);
digit += 1;
}
let value = (id[0] - b'0') as usize * 1000
+ (id[1] - b'0') as usize * 100
+ (id[2] - b'0') as usize * 10
+ (id[3] - b'0') as usize;
assert!(
value == index + 1,
"migration ids run contiguously from 0001 in apply order: a gap, a duplicate, or an entry listed out of order"
);
index += 1;
}
}
#[must_use]
pub fn migration_checksum(sql: &str) -> String {
use sha2::{Digest, Sha256};
let digest = Sha256::digest(sql.as_bytes());
let mut hex = String::with_capacity(64);
for byte in digest {
use std::fmt::Write as _;
let _ = write!(hex, "{byte:02x}");
}
hex
}
#[must_use]
pub fn is_idempotent_sql(sql: &str) -> bool {
let sql = strip_sql_comments(sql).to_ascii_lowercase();
sql.contains("if not exists") || sql.contains("if exists") || sql.contains("or replace")
}
fn strip_sql_comments(sql: &str) -> String {
let mut out = String::with_capacity(sql.len());
let mut rest = sql;
while let Some(at) = rest.find("--").into_iter().chain(rest.find("/*")).min() {
out.push_str(&rest[..at]);
let tail = &rest[at..];
if tail.starts_with("--") {
rest = tail.find('\n').map_or("", |end| &tail[end..]);
} else {
rest = tail.find("*/").map_or("", |end| &tail[end + 2..]);
}
}
out.push_str(rest);
out
}
#[must_use]
pub fn migration_missing_guard(key: &str) -> String {
format!(
"migration `{key}` is marked transactional: false, so it runs outside a transaction and \
must be idempotent: give the SQL an `IF NOT EXISTS` or `OR REPLACE` guard \
(RECONCILIATION.md, section 4)"
)
}
#[must_use]
pub fn migration_edited(key: &str, recorded: &str, found: &str) -> String {
format!(
"migration `{key}` was already applied to this database with different SQL \
(recorded sha256 {}…, embedded {}…). Applied migrations are never edited: \
revert the change and write a new migration instead \
(docs/MODULE-AUTHORING.md, step 3)",
&recorded[..recorded.len().min(12)],
&found[..found.len().min(12)]
)
}
#[derive(Debug, Clone)]
pub struct Migrations {
pub sqlite: &'static [SqlMigration],
pub postgres: &'static [SqlMigration],
}
impl Migrations {
pub const EMPTY: Migrations = Migrations {
sqlite: &[],
postgres: &[],
};
pub const fn sqlite(migrations: &'static [SqlMigration]) -> Self {
Migrations {
sqlite: migrations,
postgres: &[],
}
}
}
impl Default for Migrations {
fn default() -> Self {
Self::EMPTY
}
}
pub struct ModuleContext {
pub ports: Ports,
pub config: Arc<dyn Config>,
pub events: EventBus,
pub templates: Arc<TemplateRegistry>,
pub venture: Arc<Venture>,
pub unprotected_writes_accepted: bool,
pub personal_data: std::sync::Arc<crate::PersonalDataCatalog>,
pub ui_mounted: bool,
pub scheduled: Arc<ScheduledBudget>,
}
pub trait Module: Send + Sync + 'static {
fn name(&self) -> &'static str;
fn version(&self) -> &'static str;
fn harness_api(&self) -> u32 {
HARNESS_API
}
fn requires(&self) -> &'static [Port];
fn optional(&self) -> &'static [Port] {
&[]
}
fn depends_on(&self) -> &'static [&'static str] {
&[]
}
fn tables(&self) -> &'static [&'static str] {
&[]
}
fn actor_kinds(&self) -> &'static [&'static str] {
&[]
}
fn personal_data(&self) -> &'static [crate::PersonalDataSet] {
&[]
}
fn emits(&self) -> &'static [&'static str] {
&[]
}
fn public_writes(&self) -> bool {
false
}
fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
crate::route_policy::RoutePolicy::HumanForm
}
fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
crate::route_policy::SignatureVerification::Payments
}
fn migrations(&self) -> Migrations;
fn validate_config(&self, cfg: &dyn Config) -> Result<(), ConfigError>;
fn max_body_bytes(&self, _cfg: &dyn Config) -> usize {
crate::http::MAX_BODY_BYTES
}
fn streaming_routes(&self) -> &'static [crate::StreamRoute] {
&[]
}
fn max_blob_object_bytes(&self) -> u64 {
crate::ports::MAX_BLOB_BYTES as u64
}
fn self_check(&self) -> Vec<String> {
Vec::new()
}
fn router(&self, ctx: ModuleContext) -> axum::Router;
fn well_known(&self) -> Option<axum::Router> {
None
}
fn surface(&self) -> Surface {
Surface::none()
}
fn events(&self) -> Vec<(EventName, EventHandler)> {
Vec::new()
}
fn scheduled<'a>(
&'a self,
ctx: &'a ModuleContext,
cron: &'a str,
) -> BoxFuture<'a, Result<(), AnyError>> {
let _ = (ctx, cron);
Box::pin(async { Ok(()) })
}
}
#[cfg(test)]
mod reconciliation_guard_tests {
use super::{SqlMigration, is_idempotent_sql, migration_missing_guard};
#[test]
fn a_migration_is_transactional_unless_marked_otherwise() {
let migration = SqlMigration::new("0001", "init", "CREATE TABLE t (id TEXT PRIMARY KEY);");
assert!(migration.transactional);
assert!(!migration.clone().non_transactional().transactional);
}
#[test]
fn guarded_ddl_is_idempotent_plain_ddl_is_not() {
assert!(is_idempotent_sql(
"CREATE INDEX CONCURRENTLY IF NOT EXISTS i ON t (c)"
));
assert!(is_idempotent_sql(
"CREATE OR REPLACE FUNCTION f() RETURNS void AS $$ BEGIN END $$;"
));
assert!(is_idempotent_sql("DROP INDEX CONCURRENTLY IF EXISTS i"));
assert!(!is_idempotent_sql("CREATE INDEX CONCURRENTLY i ON t (c)"));
assert!(is_idempotent_sql("create index if not exists i on t (c)"));
assert!(!is_idempotent_sql(
"-- we could use IF NOT EXISTS here but did not\nCREATE INDEX CONCURRENTLY i ON t (c)"
));
assert!(!is_idempotent_sql(
"/* IF NOT EXISTS is unavailable on this engine */ CREATE INDEX i ON t (c)"
));
assert!(is_idempotent_sql(
"-- concurrently, so it must be idempotent\nCREATE INDEX CONCURRENTLY IF NOT EXISTS i ON t (c)"
));
}
#[test]
fn the_missing_guard_message_names_the_migration_and_the_rule() {
let message = migration_missing_guard("waitlist/0005");
assert!(message.contains("waitlist/0005"), "{message}");
assert!(message.contains("IF NOT EXISTS"), "{message}");
}
}