Archived
james.maes
9bd782efed
Uncommitted state at the 2026-10-02 owner pause. Full verify against foundation d0a11bf passes 566 tests.
671 lines
47 KiB
Java
671 lines
47 KiB
Java
/* Copyright (C) 2026 KofTwentyTwo */
|
|
package com.kof22.carlai;
|
|
|
|
import javax.sql.DataSource;
|
|
|
|
import java.math.BigDecimal;
|
|
import java.net.URL;
|
|
import java.net.URLClassLoader;
|
|
import java.nio.file.Files;
|
|
import java.nio.file.Path;
|
|
import java.security.MessageDigest;
|
|
import java.sql.SQLException;
|
|
import java.time.Clock;
|
|
import java.time.Duration;
|
|
import java.time.Instant;
|
|
import java.time.ZoneOffset;
|
|
import java.util.Collections;
|
|
import java.util.Enumeration;
|
|
import java.util.HexFormat;
|
|
import java.util.LinkedHashMap;
|
|
import java.util.List;
|
|
import java.util.Map;
|
|
import java.util.Properties;
|
|
import java.util.UUID;
|
|
import java.util.concurrent.TimeUnit;
|
|
import java.util.zip.ZipFile;
|
|
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import com.kof22.agentadmin.bootstrap.DatabaseBootstrap;
|
|
import com.kof22.agentcore.deployment.RuntimeDatabaseRole;
|
|
import com.kof22.agentcore.store.AgentLiquibase;
|
|
import com.kof22.agentcore.store.AgentMigrations;
|
|
import com.kof22.carlai.domain.CarlService;
|
|
import com.kof22.carlai.domain.TableExports;
|
|
import org.flywaydb.core.Flyway;
|
|
import org.postgresql.ds.PGSimpleDataSource;
|
|
import org.testcontainers.containers.PostgreSQLContainer;
|
|
|
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
|
import static org.junit.jupiter.api.Assertions.assertFalse;
|
|
import static org.junit.jupiter.api.Assertions.assertThrows;
|
|
import static org.junit.jupiter.api.Assertions.assertTrue;
|
|
|
|
/** Disposable actual Carl PostgreSQL migration gate; never accepts an external database URL. */
|
|
public final class CarlLiquibaseQualification
|
|
{
|
|
private static final String EXPORT_SQL = "V62__protected_table_exports.sql";
|
|
private static final UUID EXPORT_ID = UUID.fromString("00000000-0000-0000-0000-000000000005");
|
|
private static final String EXPORT_READER_COLUMNS = "id,principal,title,source_table,selection_scope,as_of,row_count,byte_count,content_hash,text_policy,household_changed";
|
|
private static final String DOCUMENT_SQL = "V61__household_documents.sql";
|
|
private static final String ORIGINAL = "Synthetic retained original. Untrusted historical evidence.";
|
|
private static final String AT = "2026-10-01T00:00:00Z";
|
|
private static final ObjectMapper JSON = new ObjectMapper();
|
|
private static final String HEADER = "<?xml version='1.0'?><databaseChangeLog xmlns='http://www.liquibase.org/xml/ns/dbchangelog' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' xsi:schemaLocation='http://www.liquibase.org/xml/ns/dbchangelog http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.33.xsd'>";
|
|
|
|
private CarlLiquibaseQualification() {}
|
|
|
|
/** Qualifies real disposable PostgreSQL, canonical domain transitions, roles and explicit recovery. */
|
|
public static void main(String[] args) throws Exception
|
|
{
|
|
if(args[0].equals("migrate-native"))
|
|
{
|
|
var properties = new Properties();
|
|
try(var reader = Files.newBufferedReader(Path.of(args[1]))) { properties.load(reader); }
|
|
try(var loader = new QualificationLoader(Path.of(args[2]), Thread.currentThread().getContextClassLoader()))
|
|
{
|
|
AgentLiquibase.migrate(source(properties, "db"), loader);
|
|
}
|
|
return;
|
|
}
|
|
Path report = Path.of(args[0]);
|
|
try(var database = new PostgreSQLContainer<>("postgres:16-alpine"))
|
|
{
|
|
database.start();
|
|
ClassLoader packaged = Thread.currentThread().getContextClassLoader();
|
|
var domainSql = sqlHashes(packaged);
|
|
boolean packagedMatch = verifyPackaged(domainSql);
|
|
int latestVersion = domainSql.keySet().stream().mapToInt(CarlLiquibaseQualification::version).max().orElseThrow();
|
|
int beforeDocuments = domainSql.keySet().stream().mapToInt(CarlLiquibaseQualification::version).filter(version -> version < version(DOCUMENT_SQL)).max().orElseThrow();
|
|
int expectedHistory = domainSql.size() + 8;
|
|
pendingDocuments(database, beforeDocuments, expectedHistory);
|
|
pendingExports(database, version(DOCUMENT_SQL), expectedHistory);
|
|
var retained = provision(database, "carl_retained", "flyway");
|
|
var retainedSource = source(retained, "db");
|
|
seed(retainedSource);
|
|
seedDocument(retainedSource);
|
|
seedExport(retainedSource);
|
|
assertExport(retainedSource);
|
|
assertDocument(retainedSource);
|
|
Map<String, String> dataBefore = records(retainedSource);
|
|
Map<String, String> legacyBefore = histories(retainedSource);
|
|
execute(admin(database), "CREATE DATABASE carl_retained_clone TEMPLATE carl_retained OWNER carl_retained_migrator");
|
|
// PostgreSQL clones table grants but creates fresh database ACLs. Restore isolated ACLs explicitly.
|
|
execute(admin(database), "REVOKE ALL ON DATABASE carl_retained_clone FROM PUBLIC");
|
|
execute(admin(database), "GRANT CONNECT ON DATABASE carl_retained_clone TO carl_retained_runtime,carl_retained_reader");
|
|
var clone = copy(retained);
|
|
clone.setProperty("db.url", url(database, "carl_retained_clone"));
|
|
clone.setProperty("migration.engine", "liquibase");
|
|
run("migrate", clone);
|
|
var cloneSource = source(clone, "db");
|
|
assertEquals(dataBefore, records(cloneSource), "Retained Monarch/privacy records must be unchanged");
|
|
assertEquals(legacyBefore, histories(cloneSource), "Independent Flyway histories must stay byte-identical");
|
|
assertEquals(expectedHistory, number(cloneSource, "SELECT count(*) FROM agent_databasechangelog"));
|
|
assertEquals("MARK_RAN", rows(cloneSource, "SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/" + DOCUMENT_SQL + "'").trim(), "Already validated document SQL must be adopted without reexecution");
|
|
assertEquals("MARK_RAN", rows(cloneSource, "SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/" + EXPORT_SQL + "'").trim(), "Already validated export SQL must be adopted without reexecution");
|
|
assertEquals(domainSql.size(), number(cloneSource, "SELECT count(*) FROM agent_databasechangelog WHERE filename LIKE 'db/migration/%'"));
|
|
Map<String, String> completeBefore = records(cloneSource);
|
|
String engineBefore = rows(cloneSource, "SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted");
|
|
run("migrate", clone);
|
|
run("migrate", clone);
|
|
assertEquals(completeBefore, records(cloneSource));
|
|
assertEquals(engineBefore, rows(cloneSource, "SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted"));
|
|
|
|
Properties fresh;
|
|
try { fresh = provision(database, "carl_fresh", "liquibase"); }
|
|
catch(IllegalStateException failure)
|
|
{
|
|
// Public SQL errors from this owned disposable server; never inspect a private/live server.
|
|
database.getLogs().lines().filter(line -> line.contains("ERROR:")).forEach(System.err::println);
|
|
System.err.println("Last applied public SQL: " + rows(source(url(database, "carl_fresh"), database.getUsername(), database.getPassword()),
|
|
"SELECT filename FROM agent_databasechangelog ORDER BY orderexecuted DESC LIMIT 1").trim());
|
|
throw failure;
|
|
}
|
|
var freshSource = source(fresh, "db");
|
|
seed(freshSource);
|
|
seedDocument(freshSource);
|
|
seedExport(freshSource);
|
|
assertExport(freshSource);
|
|
assertExport(cloneSource);
|
|
assertDocument(freshSource);
|
|
assertDocument(cloneSource);
|
|
assertEquals(schema(cloneSource), schema(freshSource), "Actual canonical fresh and retained domain schemas match");
|
|
assertEquals(dataBefore, records(freshSource), "The same Monarch rows, document originals and privacy state persist on both paths");
|
|
assertEquals(expectedHistory, number(freshSource, "SELECT count(*) FROM agent_databasechangelog"));
|
|
assertEquals(0, number(freshSource, "SELECT count(*) FROM pg_tables WHERE schemaname='public' AND tablename IN ('flyway_schema_history','agent_domain_schema_history')"));
|
|
triggerBehavior(cloneSource);
|
|
triggerBehavior(freshSource);
|
|
System.out.println("CARL_LIQUIBASE_STAGE=fresh-retained-repeat");
|
|
|
|
checksumRejection(clone, cloneSource, dataBefore);
|
|
permissions(clone);
|
|
try { permissions(fresh); }
|
|
catch(IllegalStateException failure)
|
|
{
|
|
database.getLogs().lines().filter(line -> line.contains("ERROR:")).forEach(System.err::println);
|
|
throw failure;
|
|
}
|
|
try(var connection = source(clone, "runtime").getConnection(); var statement = connection.createStatement())
|
|
{
|
|
try(var rows = statement.executeQuery("SELECT has_database_privilege(current_user,'carl_fresh','CONNECT')"))
|
|
{
|
|
rows.next();
|
|
assertFalse(rows.getBoolean(1), "Other environment roles cannot connect to this database");
|
|
}
|
|
}
|
|
System.out.println("CARL_LIQUIBASE_STAGE=checksums-role-isolation");
|
|
|
|
Path overrides = Files.createTempDirectory(report, "qualification-only-changelog-");
|
|
Path nativeLog = overrides.resolve(AgentLiquibase.DOMAIN_CHANGELOG);
|
|
Files.createDirectories(nativeLog.getParent());
|
|
String upgrade = "<changeSet id='ci-new-carl-ddl-data' author='carl-ci'><preConditions onFail='HALT'><tableExists tableName='carl_account'/><not><columnExists tableName='carl_account' columnName='migration_probe'/></not></preConditions><sql>ALTER TABLE carl_account ADD COLUMN migration_probe NUMERIC(20,4); UPDATE carl_account SET migration_probe=12.34;</sql></changeSet>";
|
|
Files.writeString(nativeLog, HEADER + upgrade + "</databaseChangeLog>");
|
|
try(var loader = new QualificationLoader(overrides, packaged))
|
|
{
|
|
AgentLiquibase.migrate(cloneSource, loader);
|
|
AgentLiquibase.migrate(freshSource, loader);
|
|
AgentLiquibase.migrate(cloneSource, loader);
|
|
assertEquals(schema(cloneSource), schema(freshSource));
|
|
assertEquals(records(cloneSource), records(freshSource));
|
|
assertEquals(new BigDecimal("12.3400"), decimal(cloneSource, "SELECT migration_probe FROM carl_account"));
|
|
assertEquals(expectedHistory + 1, number(cloneSource, "SELECT count(*) FROM agent_databasechangelog"));
|
|
interrupted(database, clone, cloneSource, nativeLog, upgrade, expectedHistory);
|
|
}
|
|
System.out.println("CARL_LIQUIBASE_STAGE=actual-interruption-explicit-recovery");
|
|
var result = new LinkedHashMap<String, Object>();
|
|
result.put("status", "PASS");
|
|
for(String gate : List.of("fresh", "retained", "repeat", "checksumRejection", "permissionIsolation", "interruptionRetry")) result.put(gate, true);
|
|
result.put("data", "DISPOSABLE_SYNTHETIC_ONLY");
|
|
result.put("consumerDomainMigrationCount", domainSql.size());
|
|
result.put("latestConsumerDomainVersion", latestVersion);
|
|
result.put("preDocumentDomainVersion", beforeDocuments);
|
|
result.put("pendingDocumentMigration", true);
|
|
result.put("retainedDocumentOriginalAndMetadata", true);
|
|
result.put("documentSqlSha256", domainSql.get(DOCUMENT_SQL));
|
|
result.put("exportSqlSha256", domainSql.get(EXPORT_SQL));
|
|
result.put("pendingDocumentAndExportMigrations", true);
|
|
result.put("documentAppliedExportPendingTransition", true);
|
|
result.put("retainedCanonicalCsvAndHistory", true);
|
|
result.put("exportRawReaderDenied", true);
|
|
result.put("coreVersions", 8);
|
|
result.put("retainedClone", true);
|
|
result.put("sourceSqlSha256", domainSql);
|
|
result.put("packagedDistributionMatched", packagedMatch);
|
|
result.put("foundationCoreSha256", hash(Files.readAllBytes(Path.of(AgentLiquibase.class.getProtectionDomain().getCodeSource().getLocation().toURI()))));
|
|
result.put("retainedDataSha256", hash(JSON.writeValueAsBytes(dataBefore)));
|
|
result.put("manualLockRecovery", "Worker finished and zero database sessions verified before explicit operator reset; no automatic unlock");
|
|
result.put("qualificationOnlyChangesets", "New DDL/data and interrupted observation execute only in disposable test changelog overrides; not production schema");
|
|
result.put("deployment", "NOT_QUALIFIED");
|
|
Files.writeString(report.resolve("report.json"), JSON.writerWithDefaultPrettyPrinter().writeValueAsString(result) + "\n");
|
|
}
|
|
}
|
|
|
|
private static Properties provision(PostgreSQLContainer<?> database, String name, String engine) throws Exception
|
|
{
|
|
return provision(database, name, engine, null);
|
|
}
|
|
|
|
private static Properties provision(PostgreSQLContainer<?> database, String name, String engine, Integer target) throws Exception
|
|
{
|
|
execute(admin(database), "CREATE DATABASE " + name);
|
|
Properties properties = new Properties();
|
|
properties.setProperty("db.url", url(database, name));
|
|
properties.setProperty("db.username", database.getUsername());
|
|
properties.setProperty("db.password", database.getPassword());
|
|
properties.setProperty("migration.engine", engine);
|
|
for(String role : List.of("migration", "runtime", "qqq.reader"))
|
|
{
|
|
properties.setProperty(role + ".username", name + "_" + (role.equals("migration") ? "migrator" : role.equals("runtime") ? "runtime" : "reader"));
|
|
properties.setProperty(role + ".password", "synthetic-qualification-" + role);
|
|
}
|
|
properties.setProperty("qqq.reader.columns.carl_account_view", "id,principal,title,kind,currency,review_state");
|
|
if(target == null)
|
|
{
|
|
properties.setProperty("qqq.reader.columns.carl_document_view", "id,principal,title,source_identity,source_type,document_date,as_of_date,state,content_hash,byte_count,extraction_status");
|
|
properties.setProperty("qqq.reader.columns.carl_table_export_view", EXPORT_READER_COLUMNS);
|
|
run("provision", properties);
|
|
}
|
|
else
|
|
{
|
|
run("prepare-restore", properties);
|
|
}
|
|
properties.setProperty("db.username", properties.getProperty("migration.username"));
|
|
properties.setProperty("db.password", properties.getProperty("migration.password"));
|
|
if(target != null)
|
|
{
|
|
var source = source(properties, "db");
|
|
// Test-only explicit target preserves the two canonical SQL histories without runtime DDL.
|
|
Flyway.configure().dataSource(source).locations(AgentMigrations.CORE_LOCATION).table(AgentMigrations.CORE_HISTORY)
|
|
.createSchemas(false).validateMigrationNaming(true).cleanDisabled(true).load().migrate();
|
|
Flyway.configure().dataSource(source).locations(AgentMigrations.DOMAIN_LOCATION).table(AgentMigrations.DOMAIN_HISTORY)
|
|
.baselineOnMigrate(true).baselineVersion("0").target(target.toString())
|
|
.createSchemas(false).validateMigrationNaming(true).cleanDisabled(true).load().migrate();
|
|
assertEquals(target.intValue(), number(source, "SELECT max(version::integer) FROM agent_domain_schema_history"));
|
|
run("grant-access", properties);
|
|
}
|
|
return properties;
|
|
}
|
|
|
|
private static void pendingDocuments(PostgreSQLContainer<?> database, int beforeDocuments, int expectedHistory) throws Exception
|
|
{
|
|
var properties = provision(database, "carl_pre_documents", "flyway", beforeDocuments);
|
|
var source = source(properties, "db");
|
|
seed(source);
|
|
assertEquals("f", rows(source, "SELECT to_regclass('carl_document') IS NOT NULL").trim());
|
|
var retained = records(source);
|
|
var flyway = histories(source);
|
|
properties.setProperty("migration.engine", "liquibase");
|
|
run("migrate", properties);
|
|
assertEquals("t", rows(source, "SELECT to_regclass('carl_document') IS NOT NULL").trim(), "Canonical pending document SQL must actually create its protected schema");
|
|
assertEquals("t", rows(source, "SELECT to_regclass('carl_table_export') IS NOT NULL").trim(), "Canonical pending V62 SQL must actually create its protected CSV schema");
|
|
properties.setProperty("qqq.reader.columns.carl_document_view", "id,principal,title,source_identity,source_type,document_date,as_of_date,state,content_hash,byte_count,extraction_status");
|
|
properties.setProperty("qqq.reader.columns.carl_table_export_view", EXPORT_READER_COLUMNS);
|
|
run("grant-access", properties);
|
|
var after = records(source);
|
|
for(var entry : retained.entrySet()) assertEquals(entry.getValue(), after.get(entry.getKey()), "Retained pre-document table " + entry.getKey());
|
|
assertEquals(flyway, histories(source), "Pending SQL must preserve both independent Flyway histories");
|
|
assertEquals(expectedHistory, number(source, "SELECT count(*) FROM agent_databasechangelog"));
|
|
assertEquals("EXECUTED", rows(source, "SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/" + DOCUMENT_SQL + "'").trim());
|
|
assertEquals(1, number(source, "SELECT count(*) FROM agent_databasechangelog WHERE filename='db/migration/" + DOCUMENT_SQL + "'"));
|
|
assertEquals("EXECUTED", rows(source, "SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/" + EXPORT_SQL + "'").trim());
|
|
seedDocument(source);
|
|
seedExport(source);
|
|
assertExport(source);
|
|
var complete = records(source);
|
|
String applied = rows(source, "SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted");
|
|
run("migrate", properties);
|
|
assertEquals(complete, records(source));
|
|
assertEquals(applied, rows(source, "SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted"));
|
|
assertEquals(flyway, histories(source));
|
|
assertDocument(source);
|
|
permissions(properties);
|
|
System.out.println("CARL_LIQUIBASE_STAGE=pre-document-pending-original-retained");
|
|
}
|
|
|
|
private static void pendingExports(PostgreSQLContainer<?> database, int beforeExports, int expectedHistory) throws Exception
|
|
{
|
|
var properties=provision(database,"carl_pre_exports","flyway",beforeExports);
|
|
var source=source(properties,"db");
|
|
seed(source);
|
|
seedDocument(source);
|
|
assertDocument(source);
|
|
assertEquals("f",rows(source,"SELECT to_regclass('carl_table_export') IS NOT NULL").trim());
|
|
var retained=records(source);
|
|
var flyway=histories(source);
|
|
properties.setProperty("migration.engine","liquibase");
|
|
run("migrate",properties);
|
|
assertEquals("t",rows(source,"SELECT to_regclass('carl_table_export') IS NOT NULL").trim());
|
|
assertEquals("MARK_RAN",rows(source,"SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/"+DOCUMENT_SQL+"'").trim());
|
|
assertEquals("EXECUTED",rows(source,"SELECT exectype FROM agent_databasechangelog WHERE filename='db/migration/"+EXPORT_SQL+"'").trim());
|
|
assertEquals(expectedHistory,number(source,"SELECT count(*) FROM agent_databasechangelog"));
|
|
for(var entry:retained.entrySet()) assertEquals(entry.getValue(),records(source).get(entry.getKey()),"Retained document-era table "+entry.getKey());
|
|
assertEquals(flyway,histories(source));
|
|
seedExport(source);
|
|
assertExport(source);
|
|
assertDocument(source);
|
|
properties.setProperty("qqq.reader.columns.carl_document_view","id,principal,title,source_identity,source_type,document_date,as_of_date,state,content_hash,byte_count,extraction_status");
|
|
properties.setProperty("qqq.reader.columns.carl_table_export_view",EXPORT_READER_COLUMNS);
|
|
run("grant-access",properties);
|
|
var complete=records(source);
|
|
String applied=rows(source,"SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted");
|
|
run("migrate",properties);
|
|
assertEquals(complete,records(source));
|
|
assertEquals(applied,rows(source,"SELECT row_to_json(t)::text FROM agent_databasechangelog t ORDER BY orderexecuted"));
|
|
assertEquals(flyway,histories(source));
|
|
permissions(properties);
|
|
System.out.println("CARL_LIQUIBASE_STAGE=document-applied-export-pending-retained");
|
|
}
|
|
|
|
private static TableExports exports(DataSource source)
|
|
{
|
|
var clock=Clock.fixed(Instant.parse(AT),ZoneOffset.UTC);
|
|
return new TableExports(new CarlService(source,clock),Map.of("carlDocuments",List.of(new TableExports.Column("id",false),new TableExports.Column("title",true),new TableExports.Column("state",true),new TableExports.Column("document_date",false),new TableExports.Column("as_of_date",false))),clock);
|
|
}
|
|
|
|
private static void seedExport(DataSource source) throws Exception
|
|
{
|
|
// Canonical production generator supplies all CSV/digest/selection fields; only fixture creation time is fixed. //
|
|
exports(source).generate("migration-owner",EXPORT_ID,"carlDocuments",TableExports.Scope.SELECTED,List.of("102"));
|
|
var fixture=(com.fasterxml.jackson.databind.node.ObjectNode)JSON.readTree(rows(source,"SELECT row_to_json(e)::text FROM carl_table_export e WHERE id='"+EXPORT_ID+"'").trim());
|
|
fixture.put("created_at",AT);
|
|
try(var connection=source.getConnection();var remove=connection.prepareStatement("DELETE FROM carl_table_export WHERE id=?");var insert=connection.prepareStatement("INSERT INTO carl_table_export SELECT * FROM json_populate_record(NULL::carl_table_export,?::json)"))
|
|
{
|
|
connection.setAutoCommit(false);
|
|
remove.setObject(1,EXPORT_ID);assertEquals(1,remove.executeUpdate());
|
|
insert.setString(1,JSON.writeValueAsString(fixture));assertEquals(1,insert.executeUpdate());
|
|
connection.commit();
|
|
}
|
|
}
|
|
|
|
private static void assertExport(DataSource source) throws Exception
|
|
{
|
|
byte[] bytes=exports(source).load("migration-owner",EXPORT_ID);
|
|
assertEquals(hash(bytes),rows(source,"SELECT content_hash FROM carl_table_export WHERE id='"+EXPORT_ID+"'").trim());
|
|
String csv=new String(bytes,java.nio.charset.StandardCharsets.UTF_8);
|
|
assertTrue(csv.contains("EXPORT_METADATA,"+AT+",carlDocuments,SELECTED,"+TableExports.TEXT_POLICY));
|
|
assertTrue(csv.contains("102,Synthetic retained household document,SUPPLIED_UNVERIFIED,,2020-01-02"));
|
|
assertFalse(csv.contains("principal"));
|
|
assertThrows(SecurityException.class,()->exports(source).load("migration-denied",EXPORT_ID));
|
|
assertEquals("carlDocuments:SELECTED:1:"+TableExports.TEXT_POLICY,rows(source,"SELECT source_table||':'||selection_scope||':'||row_count||':'||text_policy FROM carl_table_export_view WHERE principal='migration-owner' AND id='"+EXPORT_ID+"'").trim());
|
|
assertEquals(0,number(source,"SELECT count(*) FROM carl_table_export_view WHERE principal='migration-denied'"));
|
|
try(var connection=source.getConnection();var statement=connection.createStatement())
|
|
{
|
|
assertThrows(SQLException.class,()->statement.execute("UPDATE carl_table_export SET content='forged'::bytea"));
|
|
}
|
|
}
|
|
|
|
private static void seedDocument(DataSource source) throws Exception
|
|
{
|
|
String contents = HexFormat.of().formatHex(ORIGINAL.getBytes(java.nio.charset.StandardCharsets.UTF_8));
|
|
String digest = hash(ORIGINAL.getBytes(java.nio.charset.StandardCharsets.UTF_8));
|
|
try(var connection = source.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
connection.setAutoCommit(false);
|
|
for(String sql : List.of(
|
|
"INSERT INTO carl_record(id,household_id,owner_id,domain,visibility,title,evidence,created_at) VALUES(102,1,1,'FINANCE','PRIVATE','Synthetic retained household document','Synthetic supplied provenance','" + AT + "')",
|
|
"INSERT INTO carl_upload(reference,member_id,contents,created_at) VALUES('synthetic-document-upload',1,decode('" + contents + "','hex'),'" + AT + "')",
|
|
"INSERT INTO carl_document_review(id,member_id,permission_revision,upload_reference,input_digest,source_digest,original_name,content_hash,title,visibility,source_identity,source_type,document_date,as_of_date,provenance,extraction_status,extracted_text,created_at) SELECT '00000000-0000-0000-0000-000000000003',1,permission_revision,'synthetic-document-upload','" + digest + "','" + digest + "','synthetic-retained.txt','" + digest + "','Synthetic retained household document','PRIVATE','synthetic://supplied-retained-note','HISTORICAL_NOTE',NULL,'2020-01-02','Synthetic supplied provenance','TEXT_EXTRACTED','" + ORIGINAL + "','" + AT + "' FROM carl_household WHERE id=1",
|
|
"INSERT INTO carl_document(record_id,source_identity,source_type,document_date,as_of_date,source_digest,original_name,media_type,content_hash,original_content,extraction_status,extracted_text,review_id) VALUES(102,'synthetic://supplied-retained-note','HISTORICAL_NOTE',NULL,'2020-01-02','" + digest + "','synthetic-retained.txt','text/plain','" + digest + "',decode('" + contents + "','hex'),'TEXT_EXTRACTED','" + ORIGINAL + "','00000000-0000-0000-0000-000000000003')",
|
|
"INSERT INTO carl_document_download(id,document_id,requester_id,permission_revision) SELECT '00000000-0000-0000-0000-000000000004',102,1,permission_revision FROM carl_household WHERE id=1")) statement.execute(sql);
|
|
connection.commit();
|
|
}
|
|
}
|
|
|
|
private static void assertDocument(DataSource source) throws Exception
|
|
{
|
|
assertEquals(HexFormat.of().formatHex(ORIGINAL.getBytes(java.nio.charset.StandardCharsets.UTF_8)), rows(source, "SELECT encode(original_content,'hex') FROM carl_document WHERE record_id=102").trim());
|
|
assertEquals(hash(ORIGINAL.getBytes(java.nio.charset.StandardCharsets.UTF_8)), rows(source, "SELECT content_hash FROM carl_document WHERE record_id=102").trim());
|
|
assertEquals("synthetic://supplied-retained-note:HISTORICAL_NOTE:UNKNOWN:2020-01-02:SUPPLIED_UNVERIFIED:synthetic-retained.txt:text/plain", rows(source, "SELECT source_identity||':'||source_type||':'||coalesce(document_date::text,'UNKNOWN')||':'||as_of_date::text||':'||state||':'||original_name||':'||media_type FROM carl_document_view WHERE principal='migration-owner' AND id=102").trim());
|
|
assertEquals(0, number(source, "SELECT count(*) FROM carl_document_view WHERE principal='migration-denied'"));
|
|
try(var connection = source.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
assertThrows(SQLException.class, () -> statement.execute("UPDATE carl_document SET extracted_text='forged' WHERE record_id=102"));
|
|
assertThrows(SQLException.class, () -> statement.execute("UPDATE carl_document_review SET source_identity='forged' WHERE id='00000000-0000-0000-0000-000000000003'"));
|
|
}
|
|
}
|
|
|
|
private static void seed(DataSource source) throws Exception
|
|
{
|
|
try(var connection = source.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
connection.setAutoCommit(false);
|
|
for(String sql : List.of(
|
|
"INSERT INTO carl_household(id,name,display_zone) VALUES(1,'Synthetic Migration Household','America/Chicago')",
|
|
"INSERT INTO carl_member(id,household_id,principal,label,can_manage) VALUES(1,1,'migration-owner','Owner',true),(2,1,'migration-denied','Denied',false)",
|
|
"INSERT INTO carl_permission(member_id,domain,details) VALUES(1,'FINANCE',true),(2,'FINANCE',false)",
|
|
"INSERT INTO carl_record(id,household_id,owner_id,domain,visibility,title,evidence,created_at) VALUES(100,1,1,'FINANCE','PRIVATE','Fictional unreviewed source','Synthetic migration evidence','" + AT + "'),(101,1,1,'FINANCE','PRIVATE','Fictional source transaction','Synthetic migration evidence','" + AT + "')",
|
|
"INSERT INTO carl_account(record_id,kind,institution,currency,liquid,ownership_share,review_state) VALUES(100,'UNCLASSIFIED','Fictional institution','USD',null,null,'NEEDS_REVIEW')",
|
|
"INSERT INTO carl_monarch_mapping(member_id,source_label,account_id,created_at) VALUES(1,'Fictional unreviewed source',100,'" + AT + "')",
|
|
"INSERT INTO carl_request(id,member_id,kind,digest,status,created_at) VALUES('00000000-0000-0000-0000-000000000001',1,'MONARCH_IMPORT','" + "a".repeat(64) + "','COMPLETE','" + AT + "')",
|
|
"INSERT INTO carl_import(id,household_id,member_id,source_name,content_digest,request_id,row_count,created_at) VALUES(1,1,1,'Synthetic Monarch export','" + "b".repeat(64) + "','00000000-0000-0000-0000-000000000001',1,'" + AT + "')",
|
|
"INSERT INTO carl_transaction(record_id,account_id,effective_date,amount,currency,source_id,classification,category,import_id) VALUES(101,100,'2026-10-01',-123.4500,'USD','synthetic-monarch-row','UNCLASSIFIED','Unknown',1)",
|
|
"INSERT INTO carl_transaction_source(household_id,external_id,transaction_id,version,payload_digest,import_id,logical_row,account_label,merchant,category,original_statement,notes,tags,owner_label,reviewed,amount,effective_date,created_at) VALUES(1,'synthetic-monarch-row',101,1,'" + "c".repeat(64) + "',1,2,'Fictional unreviewed source','Fictional merchant','Unknown','Fictional statement','Synthetic private note','', 'Unknown','false',-123.4500,'2026-10-01','" + AT + "')",
|
|
"INSERT INTO carl_balance(id,account_id,as_of,amount,basis,evidence) VALUES(1,100,'2026-10-01',-456.7800,'STATEMENT','Synthetic migration observation')",
|
|
"INSERT INTO carl_balance_source(account_id,as_of,amount,import_id,logical_row,account_label,created_at) VALUES(100,'2026-10-01',-456.7800,1,2,'Fictional unreviewed source','" + AT + "')",
|
|
"INSERT INTO carl_upload(reference,member_id,contents,created_at) VALUES('synthetic-upload',1,decode('736f75726365','hex'),'" + AT + "')",
|
|
"INSERT INTO carl_import_review(id,member_id,transaction_reference,status,result,created_at) VALUES('00000000-0000-0000-0000-000000000002',1,'synthetic-upload','COMPLETE','Synthetic retained receipt','" + AT + "')",
|
|
"INSERT INTO transcript_entry(session_key,seq,role,content,created_at) VALUES('synthetic-retained-conversation',1,'user','Synthetic protected conversation','" + AT + "')")) statement.execute(sql);
|
|
statement.execute("SELECT setval(pg_get_serial_sequence('carl_balance','id'),1,true)");
|
|
connection.commit();
|
|
}
|
|
}
|
|
|
|
private static void triggerBehavior(DataSource source) throws Exception
|
|
{
|
|
try(var connection = source.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
connection.setAutoCommit(false);
|
|
int before = number(source, "SELECT permission_revision FROM carl_household WHERE id=1");
|
|
statement.execute("INSERT INTO carl_member(id,household_id,principal,label,can_manage) VALUES(3,1,'migration-trigger-probe','Probe',false)");
|
|
statement.execute("UPDATE carl_member SET label='Updated probe' WHERE id=3");
|
|
statement.execute("DELETE FROM carl_member WHERE id=3");
|
|
try(var rows = statement.executeQuery("SELECT permission_revision FROM carl_household WHERE id=1"))
|
|
{
|
|
rows.next();
|
|
assertEquals(before + 3, rows.getInt(1), "Actual Carl PL/pgSQL trigger must handle insert, update and deletion");
|
|
}
|
|
connection.rollback();
|
|
assertEquals(before, number(source, "SELECT permission_revision FROM carl_household WHERE id=1"));
|
|
}
|
|
}
|
|
|
|
private static void checksumRejection(Properties properties, DataSource source, Map<String, String> records) throws Exception
|
|
{
|
|
int latest = number(source, "SELECT max(version::integer) FROM agent_domain_schema_history");
|
|
int checksum = number(source, "SELECT checksum FROM agent_domain_schema_history WHERE version='" + latest + "'");
|
|
execute(source, "UPDATE agent_domain_schema_history SET checksum=0 WHERE version='" + latest + "'");
|
|
assertThrows(IllegalStateException.class, () -> run("migrate", properties));
|
|
assertEquals(records, records(source));
|
|
execute(source, "UPDATE agent_domain_schema_history SET checksum=" + checksum + " WHERE version='" + latest + "'");
|
|
String checksumValue = rows(source, "SELECT md5sum FROM agent_databasechangelog WHERE filename LIKE '%V" + latest + "__%'");
|
|
execute(source, "UPDATE agent_databasechangelog SET md5sum='9:00000000000000000000000000000000' WHERE filename LIKE '%V" + latest + "__%'");
|
|
assertThrows(IllegalStateException.class, () -> run("migrate", properties));
|
|
assertEquals(records, records(source));
|
|
try(var connection = source.getConnection(); var statement = connection.prepareStatement("UPDATE agent_databasechangelog SET md5sum=? WHERE filename LIKE '%V" + latest + "__%'"))
|
|
{
|
|
statement.setString(1, checksumValue.trim());
|
|
statement.executeUpdate();
|
|
}
|
|
run("migrate", properties);
|
|
}
|
|
|
|
private static void permissions(Properties properties) throws Exception
|
|
{
|
|
DataSource runtime = source(properties, "runtime");
|
|
try(var connection = runtime.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
statement.executeQuery("SELECT count(*) FROM information_schema.tables WHERE table_name='agent_databasechangelog'").close();
|
|
}
|
|
assertTrue(AgentLiquibase.installed(runtime), "Runtime must recognize the installed migration engine");
|
|
AgentMigrations.validate(runtime);
|
|
try(var connection = runtime.getConnection(); var statement = connection.createStatement())
|
|
{
|
|
RuntimeDatabaseRole.verify(connection);
|
|
assertEquals(1, statement.executeUpdate("UPDATE carl_transaction_source SET notes=notes WHERE external_id='synthetic-monarch-row'"));
|
|
for(String sql : List.of("CREATE TABLE forbidden(id INTEGER)", "CREATE TEMP TABLE forbidden(id INTEGER)",
|
|
"CREATE ROLE forbidden", "DELETE FROM agent_databasechangelog", "UPDATE agent_databasechangeloglock SET locked=true",
|
|
"ALTER TABLE carl_account ADD COLUMN forbidden INTEGER", "SET ROLE " + properties.getProperty("migration.username")))
|
|
assertThrows(SQLException.class, () -> statement.execute(sql), sql);
|
|
assertThrows(IllegalStateException.class, () -> AgentLiquibase.migrate(runtime));
|
|
}
|
|
try(var connection = source(properties, "qqq.reader").getConnection(); var statement = connection.createStatement())
|
|
{
|
|
try(var rows = statement.executeQuery("SELECT kind,review_state FROM carl_account_view WHERE principal='migration-owner'"))
|
|
{
|
|
assertTrue(rows.next());
|
|
assertEquals("UNCLASSIFIED", rows.getString(1));
|
|
assertEquals("NEEDS_REVIEW", rows.getString(2));
|
|
}
|
|
try(var rows = statement.executeQuery("SELECT count(id) FROM carl_account_view WHERE principal='migration-denied'"))
|
|
{
|
|
rows.next();
|
|
assertEquals(0, rows.getInt(1));
|
|
}
|
|
try(var rows = statement.executeQuery("SELECT source_identity,as_of_date FROM carl_document_view WHERE principal='migration-owner' AND id=102"))
|
|
{
|
|
assertTrue(rows.next());
|
|
assertEquals("synthetic://supplied-retained-note", rows.getString(1));
|
|
assertEquals("2020-01-02", rows.getDate(2).toString());
|
|
}
|
|
try(var rows = statement.executeQuery("SELECT count(id) FROM carl_document_view WHERE principal='migration-denied'"))
|
|
{
|
|
rows.next();
|
|
assertEquals(0, rows.getInt(1));
|
|
}
|
|
try(var rows=statement.executeQuery("SELECT source_table,row_count,text_policy FROM carl_table_export_view WHERE principal='migration-owner'"))
|
|
{
|
|
assertTrue(rows.next());assertEquals("carlDocuments",rows.getString(1));assertEquals(1,rows.getInt(2));assertEquals(TableExports.TEXT_POLICY,rows.getString(3));
|
|
}
|
|
try(var rows=statement.executeQuery("SELECT count(id) FROM carl_table_export_view WHERE principal='migration-denied'"))
|
|
{
|
|
rows.next();assertEquals(0,rows.getInt(1));
|
|
}
|
|
for(String sql : List.of("SELECT content FROM carl_table_export", "SELECT requester_id FROM carl_table_export", "SELECT source_digest FROM carl_table_export_view", "UPDATE carl_table_export SET content='forged'::bytea", "SELECT original_content FROM carl_document", "SELECT extracted_text FROM carl_document", "SELECT * FROM carl_document_review", "SELECT * FROM carl_document_download", "UPDATE carl_document SET source_identity='forged'", "SELECT evidence FROM carl_account_view", "SELECT * FROM carl_transaction_source",
|
|
"SELECT * FROM agent_databasechangelog", "SELECT * FROM agent_databasechangeloglock",
|
|
"UPDATE carl_account SET review_state='CONFIRMED'", "CREATE TEMP TABLE forbidden(id INTEGER)"))
|
|
assertThrows(SQLException.class, () -> statement.execute(sql), sql);
|
|
}
|
|
}
|
|
|
|
private static void interrupted(PostgreSQLContainer<?> database, Properties properties, DataSource source, Path nativeLog, String upgrade, int expectedHistory) throws Exception
|
|
{
|
|
String insert = "<changeSet id='ci-interrupted-carl-data' author='carl-ci'><preConditions onFail='HALT'><tableExists tableName='carl_balance'/></preConditions><sql>INSERT INTO carl_balance(account_id,as_of,amount,basis,evidence) VALUES(100,'2026-10-02',12.34,'HUMAN_ASSERTION','Synthetic retry evidence'); ";
|
|
// A new immutable distribution has a new resource root, while retaining logical changeset identities.
|
|
Path nextDistribution = Files.createTempDirectory(nativeLog.getParent().getParent().getParent().getParent(), "qualification-only-upgrade-");
|
|
nativeLog = nextDistribution.resolve(AgentLiquibase.DOMAIN_CHANGELOG);
|
|
Files.createDirectories(nativeLog.getParent());
|
|
Files.writeString(nativeLog, HEADER + upgrade + insert + "SELECT pg_sleep(30);</sql></changeSet></databaseChangeLog>");
|
|
Path credentials = Files.createTempFile("carl-synthetic-native-", ".properties");
|
|
try
|
|
{
|
|
try(var output = Files.newBufferedWriter(credentials)) { properties.store(output, "Owned disposable credentials"); }
|
|
var worker = nativeWorker(credentials, nextDistribution);
|
|
Integer backend = null;
|
|
long deadline = System.nanoTime() + Duration.ofSeconds(20).toNanos();
|
|
while(backend == null && System.nanoTime() < deadline)
|
|
{
|
|
if(!worker.isAlive())
|
|
{
|
|
throw new IllegalStateException("Interrupted migration finished before transactional pause; see native-worker.log");
|
|
}
|
|
String active = rows(admin(database), "SELECT pid FROM pg_stat_activity WHERE datname='carl_retained_clone' AND query LIKE '%pg_sleep(30)%'").trim();
|
|
if(!active.isEmpty()) backend = Integer.valueOf(active); else Thread.sleep(50);
|
|
}
|
|
assertTrue(backend != null, "Actual Carl data migration must reach transactional pause");
|
|
assertEquals("t", rows(admin(database), "SELECT pg_terminate_backend(" + backend + ")").trim());
|
|
assertTrue(worker.waitFor(15, TimeUnit.SECONDS), "Terminated migration JVM must finish");
|
|
assertTrue(worker.exitValue() != 0, "Terminated migration must fail");
|
|
assertEquals(0, number(source, "SELECT count(*) FROM carl_balance WHERE evidence='Synthetic retry evidence'"));
|
|
assertEquals(expectedHistory + 1, number(source, "SELECT count(*) FROM agent_databasechangelog"));
|
|
assertEquals("t", rows(source, "SELECT locked FROM agent_databasechangeloglock WHERE id=1").trim());
|
|
assertEquals(0, number(admin(database), "SELECT count(*) FROM pg_stat_activity WHERE datname='carl_retained_clone'"));
|
|
// The worker is done and no sessions remain: explicit operator recovery only.
|
|
execute(source, "UPDATE agent_databasechangeloglock SET locked=false,lockgranted=null,lockedby=null WHERE id=1");
|
|
Files.writeString(nativeLog, HEADER + upgrade + insert + "</sql></changeSet></databaseChangeLog>");
|
|
for(int repeat = 0; repeat < 2; repeat++)
|
|
{
|
|
var retry = nativeWorker(credentials, nextDistribution);
|
|
assertTrue(retry.waitFor(30, TimeUnit.SECONDS));
|
|
assertEquals(0, retry.exitValue(), "Unapplied change retries in a fresh offline migration JVM");
|
|
}
|
|
assertEquals(1, number(source, "SELECT count(*) FROM carl_balance WHERE evidence='Synthetic retry evidence'"));
|
|
assertEquals(new BigDecimal("12.3400"), decimal(source, "SELECT amount FROM carl_balance WHERE evidence='Synthetic retry evidence'"));
|
|
assertEquals(expectedHistory + 2, number(source, "SELECT count(*) FROM agent_databasechangelog"));
|
|
assertEquals("f", rows(source, "SELECT locked FROM agent_databasechangeloglock WHERE id=1").trim());
|
|
}
|
|
finally { Files.deleteIfExists(credentials); }
|
|
}
|
|
|
|
private static Process nativeWorker(Path credentials, Path resources) throws Exception
|
|
{
|
|
return new ProcessBuilder(Path.of(System.getProperty("java.home"), "bin", "java").toString(), "-Xmx512m", "-cp", System.getProperty("java.class.path"),
|
|
CarlLiquibaseQualification.class.getName(), "migrate-native", credentials.toString(), resources.toString())
|
|
.redirectErrorStream(true).redirectOutput(ProcessBuilder.Redirect.appendTo(resources.resolve("native-worker.log").toFile())).start();
|
|
}
|
|
|
|
private static Map<String, String> records(DataSource source) throws Exception
|
|
{
|
|
var result = new LinkedHashMap<String, String>();
|
|
String tables = rows(source, "SELECT tablename FROM pg_tables WHERE schemaname='public' AND (tablename LIKE 'carl_%' OR tablename='transcript_entry') ORDER BY tablename");
|
|
for(String table : tables.lines().toList()) result.put(table, rows(source, "SELECT row_to_json(t)::text FROM " + table + " t ORDER BY row_to_json(t)::text"));
|
|
return result;
|
|
}
|
|
|
|
private static Map<String, String> histories(DataSource source) throws Exception
|
|
{
|
|
return Map.of("core", rows(source, "SELECT row_to_json(t)::text FROM flyway_schema_history t ORDER BY installed_rank"),
|
|
"domain", rows(source, "SELECT row_to_json(t)::text FROM agent_domain_schema_history t ORDER BY installed_rank"));
|
|
}
|
|
|
|
private static String schema(DataSource source) throws Exception
|
|
{
|
|
return rows(source, "SELECT table_name||':'||column_name||':'||data_type||':'||coalesce(column_default,'')||':'||is_nullable FROM information_schema.columns WHERE table_schema='public' AND table_name LIKE 'carl_%' ORDER BY table_name,ordinal_position")
|
|
+ rows(source, "SELECT viewname||':'||definition FROM pg_views WHERE schemaname='public' AND viewname LIKE 'carl_%' ORDER BY viewname");
|
|
}
|
|
|
|
private static Map<String, String> sqlHashes(ClassLoader loader) throws Exception
|
|
{
|
|
var result = new LinkedHashMap<String, String>();
|
|
var migrations = Path.of(loader.getResource("db/migration").toURI());
|
|
try(var paths = Files.list(migrations))
|
|
{
|
|
for(Path path : paths.sorted().toList()) if(path.getFileName().toString().matches("V[0-9]+__.*\\.sql")) result.put(path.getFileName().toString(), hash(Files.readAllBytes(path)));
|
|
}
|
|
assertTrue(result.containsKey("V60__unreviewed_source_accounts.sql"));
|
|
return result;
|
|
}
|
|
|
|
private static boolean verifyPackaged(Map<String, String> sql) throws Exception
|
|
{
|
|
String distribution = System.getenv("CARL_QUALIFIED_DISTRIBUTION");
|
|
if(distribution == null || distribution.isBlank()) return false;
|
|
Path core = Path.of(AgentLiquibase.class.getProtectionDomain().getCodeSource().getLocation().toURI());
|
|
assertEquals(hash(Files.readAllBytes(core)), hash(Files.readAllBytes(Path.of(distribution, "lib", core.getFileName().toString()))),
|
|
"Tested migration engine must match the packaged runtime engine");
|
|
try(var archive = new ZipFile(Path.of(distribution, "app.jar").toFile()))
|
|
{
|
|
for(var entry : sql.entrySet())
|
|
{
|
|
var resource = archive.getEntry("db/migration/" + entry.getKey());
|
|
assertTrue(resource != null, "Every actual legacy SQL resource must be packaged");
|
|
try(var input = archive.getInputStream(resource)) { assertEquals(entry.getValue(), hash(input.readAllBytes())); }
|
|
}
|
|
var nativeLog = archive.getEntry(AgentLiquibase.DOMAIN_CHANGELOG);
|
|
assertTrue(nativeLog != null, "Conventional native consumer changelog must be packaged");
|
|
try(var input = archive.getInputStream(nativeLog); var expected = Thread.currentThread().getContextClassLoader().getResourceAsStream(AgentLiquibase.DOMAIN_CHANGELOG))
|
|
{
|
|
assertEquals(hash(expected.readAllBytes()), hash(input.readAllBytes()));
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private static int version(String resource) { return Integer.parseInt(resource.substring(1, resource.indexOf("__"))); }
|
|
private static String hash(byte[] bytes) throws Exception { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); }
|
|
private static Properties copy(Properties properties) { var result = new Properties(); result.putAll(properties); return result; }
|
|
private static String url(PostgreSQLContainer<?> database, String name) { return database.getJdbcUrl().replace("/" + database.getDatabaseName(), "/" + name); }
|
|
private static DataSource admin(PostgreSQLContainer<?> database) { return source(database.getJdbcUrl(), database.getUsername(), database.getPassword()); }
|
|
private static DataSource source(Properties properties, String prefix) { return source(properties.getProperty("db.url"), properties.getProperty(prefix + ".username"), properties.getProperty(prefix + ".password")); }
|
|
private static DataSource source(String url, String user, String password) { var source = new PGSimpleDataSource(); source.setURL(url); source.setUser(user); source.setPassword(password); return source; }
|
|
private static void execute(DataSource source, String sql) throws Exception { try(var connection = source.getConnection(); var statement = connection.createStatement()) { statement.execute(sql); } }
|
|
private static int number(DataSource source, String sql) throws Exception { return Integer.parseInt(rows(source, sql).trim()); }
|
|
private static BigDecimal decimal(DataSource source, String sql) throws Exception { return new BigDecimal(rows(source, sql).trim()); }
|
|
private static String rows(DataSource source, String sql) throws Exception
|
|
{
|
|
var result = new StringBuilder();
|
|
try(var connection = source.getConnection(); var statement = connection.createStatement(); var rows = statement.executeQuery(sql))
|
|
{
|
|
while(rows.next()) result.append(rows.getString(1)).append('\n');
|
|
}
|
|
return result.toString();
|
|
}
|
|
|
|
private static void run(String operation, Properties properties) throws Exception
|
|
{
|
|
Path file = Files.createTempFile("carl-synthetic-migration-", ".properties");
|
|
try
|
|
{
|
|
try(var output = Files.newBufferedWriter(file)) { properties.store(output, "Disposable synthetic migration credentials"); }
|
|
DatabaseBootstrap.main(new String[]{operation, file.toString()});
|
|
}
|
|
finally { Files.deleteIfExists(file); }
|
|
}
|
|
|
|
private static final class QualificationLoader extends URLClassLoader
|
|
{
|
|
private final Path resources;
|
|
private QualificationLoader(Path resources, ClassLoader parent) throws Exception { super(new URL[]{resources.toUri().toURL()}, parent); this.resources = resources; }
|
|
@Override public URL getResource(String name)
|
|
{
|
|
try { if(name.equals(AgentLiquibase.DOMAIN_CHANGELOG)) return resources.resolve(name).toUri().toURL(); }
|
|
catch(java.net.MalformedURLException failure) { throw new IllegalStateException(failure); }
|
|
return super.getResource(name);
|
|
}
|
|
@Override public Enumeration<URL> getResources(String name) throws java.io.IOException
|
|
{
|
|
if(name.equals(AgentLiquibase.DOMAIN_CHANGELOG)) return Collections.enumeration(List.of(resources.resolve(name).toUri().toURL()));
|
|
return super.getResources(name);
|
|
}
|
|
}
|
|
}
|