diff --git a/.github/workflows/ci-integ-test.yml b/.github/workflows/ci-integ-test.yml index 529289cd..320430bc 100644 --- a/.github/workflows/ci-integ-test.yml +++ b/.github/workflows/ci-integ-test.yml @@ -20,7 +20,30 @@ jobs: - uses: gradle/gradle-build-action@v2 with: gradle-version: 7.5 - - run: git submodule update --init --recursive && sleep 30 && gradle build + - name: "Wait for integ Postgres to be ready" + run: | + for i in $(seq 1 30); do + if docker exec iwf-integ-postgres pg_isready -U iwf -d iwf_integ; then + echo "postgres is ready"; exit 0 + fi + echo "waiting for postgres... ($i)"; sleep 2 + done + echo "postgres did not become ready in time" >&2; exit 1 + - name: "Wait for iWF server to be ready" + run: | + # actively wait until the iWF server serves (it depends on Temporal + Elasticsearch being up), + # instead of relying on a fixed sleep which is too short for the full stack to start. + for i in $(seq 1 60); do + code=$(curl -s -o /dev/null -w "%{http_code}" -X POST http://localhost:8801/api/v1/workflow/start || echo 000) + if [ "$code" = "400" ] || [ "$code" = "200" ]; then + echo "iwf-server is ready (http $code)"; exit 0 + fi + echo "waiting for iwf-server... ($i, http $code)"; sleep 3 + done + echo "iwf-server did not become ready in time" >&2 + docker logs iwf-server 2>&1 | tail -50 || true + exit 1 + - run: git submodule update --init --recursive && gradle build - name: Dump docker logs if: always() uses: jwalton/gh-docker-logs@v2 diff --git a/README.md b/README.md index e18ba2e7..2b639333 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,11 @@ A workflow can contain any number of WorkflowStates. See more in https://github.com/indeedeng/iwf#what-is-iwf +## Features + +* [Data Attribute → Postgres sync](./docs/data-attribute-db-sync-overview.md) — automatically mirror a + data attribute to a cell of a user-owned Postgres table (load on start, sync on mutation). + ## How to build & run ### Using IntelliJ diff --git a/build.gradle b/build.gradle index b5f4659e..1f0e9a29 100644 --- a/build.gradle +++ b/build.gradle @@ -75,6 +75,10 @@ dependencies { testImplementation 'org.springframework.boot:spring-boot-starter-test' testImplementation "javax.validation:validation-api:2.0.1.Final" + // Postgres JDBC driver for the data-attribute DB-sync integration test (test-only; the core SDK + // stays driver-agnostic and depends only on javax.sql.DataSource). + testImplementation 'org.postgresql:postgresql:42.7.4' + // immutable compileOnly 'org.immutables:value:2.9.2' diff --git a/docs/data-attribute-db-sync-overview.md b/docs/data-attribute-db-sync-overview.md new file mode 100644 index 00000000..c5d8c6b0 --- /dev/null +++ b/docs/data-attribute-db-sync-overview.md @@ -0,0 +1,36 @@ +# Data Attribute → Postgres sync (overview) + +Automatically mirror a workflow **data attribute** to a cell (row + column) of a user-owned Postgres +table. The iWF data attribute stays the source of truth; the database cell is a mirror. + +- **Load on start** — when the workflow starts, the data attribute is loaded from its mapped column. +- **Sync on mutation** — whenever the data attribute is mutated in a state's `execute` method, the new + value is written back to the same column. + +```java +DataSource ds = /* your configured, pooled DataSource */; + +DataAttributeDef.create( + String.class, + "status", + DbAttributeSync.of(ds, "public.orders", "id", + (workflowId, persistence) -> new CellLocation(workflowId, "status_col"))); +``` + +## How it works (3 lines) + +1. `Client.startWorkflow` starts at a load system state that `SELECT`s each mapped column and sets the + data attribute, then transitions to your real starting state (input passes through). +2. When a state's `execute` mutates a mapped data attribute, the worker reroutes the decision through a + sync system state that `UPDATE`s the column, then continues to your original decision. +3. iWF persists the data attribute first, then the DB write happens — so a DB failure retries only the + sync step, never your `execute`. Neither system state is registered; both are worker-internal. + +## Key limitations (v1) + +- Fixed-key data attributes only (not `createByPrefix`). +- Sync fires on `execute` mutations only (not `waitUntil`). +- Load fires for workflows started through the registered `Client`. +- The mapped column stores the data attribute's JSON — use a `text`/`jsonb` column. + +See the detailed guide: [data-attribute-db-sync.md](./data-attribute-db-sync.md). diff --git a/docs/data-attribute-db-sync.md b/docs/data-attribute-db-sync.md new file mode 100644 index 00000000..a0d39217 --- /dev/null +++ b/docs/data-attribute-db-sync.md @@ -0,0 +1,164 @@ +# Data Attribute → Postgres sync (user guide) + +This feature lets you automatically mirror an iWF **data attribute** to a single cell (a row + column) +of an arbitrary, **user-owned** Postgres table: + +- **Load on start**: when a workflow starts, the data attribute's initial value is loaded from the + mapped database column. +- **Sync on mutation**: whenever the data attribute is mutated inside a state's `execute` method, the + new value is written back to the same column. + +It is entirely an **SDK-side** feature — iWF's server is not involved in your database, and you own the +database schema, credentials, and connection pooling. + +## Mental model + +The iWF data attribute is the **source of truth**; the Postgres cell is a **mirror**. + +- On start, the mirror seeds the data attribute (load). +- During execution, the data attribute drives the mirror (sync). +- iWF persists the data attribute first, then the database is updated. If the database write fails, only + the (internal) sync step is retried — your `execute` code is **not** re-run. + +## Prerequisites + +1. A Postgres table you own, with: + - a **primary key column** (text/uuid/numeric — the PK value is bound as a string parameter), and + - a **data column** of type `text` or `jsonb` (the SDK stores the data attribute's JSON there). +2. A configured `javax.sql.DataSource`. You own it — the SDK never stores credentials and never creates + connections other than via `dataSource.getConnection()`. Use any driver/pool (HikariCP, + `PGSimpleDataSource`, etc.). + +Example table: + +```sql +CREATE TABLE orders ( + id TEXT PRIMARY KEY, + status_col TEXT +); +INSERT INTO orders (id, status_col) VALUES ('order-123', '"NEW"'); +``` + +## Setup + +### 1. Build a DataSource + +```java +import org.postgresql.ds.PGSimpleDataSource; +import javax.sql.DataSource; + +PGSimpleDataSource ds = new PGSimpleDataSource(); +ds.setUrl("jdbc:postgresql://localhost:5432/mydb"); +ds.setUser("app"); +ds.setPassword(System.getenv("DB_PASSWORD")); // keep credentials out of code +// (in production prefer a pooled DataSource, e.g. HikariCP) +``` + +### 2. Declare the data attribute with a DB-sync mapping + +Use the `DataAttributeDef.create(type, key, DbAttributeSync)` overload in your workflow's +`getPersistenceSchema()`: + +```java +import io.iworkflow.core.persistence.*; + +public class OrderWorkflow implements ObjectWorkflow { + public static final String STATUS = "status"; + + @Override + public List getPersistenceSchema() { + return Arrays.asList( + DataAttributeDef.create( + String.class, + STATUS, + DbAttributeSync.of( + /* dataSource */ MyDataSources.ORDERS, + /* table */ "orders", // optionally schema-qualified: "public.orders" + /* pk column */ "id", + /* locator */ (workflowId, persistence) -> + new CellLocation(workflowId, "status_col")))); + } + + @Override + public List getWorkflowStates() { + return Arrays.asList(StateDef.startingState(new DecideState())); + } +} +``` + +**The locator** `(String workflowId, Persistence persistence) -> CellLocation` returns which cell to +load-from / sync-to: + +- `CellLocation.getPkValue()` — the primary-key **value** identifying the row (bound as a parameter). +- `CellLocation.getColumnName()` — the **data column** name. + +The table name and the primary-key column name are fixed in `DbAttributeSync.of(...)`. The locator can +compute the PK value from the `workflowId` and/or from the current `persistence` (other data attributes +already set at that point). + +### 3. Use the data attribute normally + +```java +public class DecideState implements WorkflowState { + @Override public Class getInputType() { return String.class; } + + @Override + public StateDecision execute(Context ctx, String in, CommandResults r, + Persistence persistence, Communication comm) { + String status = persistence.getDataAttribute(OrderWorkflow.STATUS, String.class); // loaded from DB + // ... business logic ... + persistence.setDataAttribute(OrderWorkflow.STATUS, "PROCESSED"); // mirrored back to DB + return StateDecision.gracefulCompleteWorkflow(status); + } +} +``` + +Start the workflow as usual with the registered `Client` — the load happens automatically: + +```java +client.startWorkflow(OrderWorkflow.class, "order-123", 3600, "start"); +``` + +## Behavior / semantics + +| Aspect | Behavior | +|---|---| +| When load fires | Once, at workflow start (via the registered `Client`). | +| When sync fires | After a state's `execute` mutates a mapped data attribute. | +| Ordering | iWF persists the data attribute first; then the DB `UPDATE` runs. | +| Retry / idempotency | The DB write is a single-row idempotent `UPDATE`; safe to retry. On failure only the internal sync step retries — your `execute` is not re-run. | +| Stored value | The data attribute's `ObjectEncoder` JSON (same bytes iWF stores). Use a `text`/`jsonb` column. | +| Multiple synced attributes | Each maps to its own cell; all load at start and each syncs when mutated. | + +## Column & type guidance + +The mapped column stores the data attribute's JSON. For a `String` attribute of value `NEW`, the column +holds the JSON string `"NEW"`. For complex objects it holds the serialized JSON object. This round-trips +any type uniformly, so use `text` or `jsonb`. Mapping to a native scalar column (e.g. an `integer` +column) is **not** supported in v1. + +## Limitations (v1) + +- **Fixed-key data attributes only** — `DataAttributeDef.createByPrefix(...)` is not supported for DB + sync (a `WorkflowDefinitionException` is thrown at registration if you try). +- **`execute` mutations only** — a data attribute mutated in `waitUntil` (and not in `execute`) is not + synced. Mutate DB-synced data attributes in `execute`. +- **Started via the registered `Client`** — starts issued directly against the server API / another SDK + with an explicit start stateId bypass the load step. + +## Troubleshooting + +| Symptom | Likely cause | +|---|---| +| Data attribute is `null` after start | Row/PK/column mismatch — the locator's PK value or column name doesn't match an existing row/column, or the column is null. | +| Column not updated after a state | The attribute was mutated in `waitUntil` (not `execute`), or the mutated key isn't the DB-mapped one. | +| `unsafe SQL identifier` error | The table / PK / data-column name isn't a plain identifier (`^[A-Za-z_][A-Za-z0-9_]*$`, optionally schema-qualified). Rename or quote-safe your identifiers. | +| Connection / pool errors | Check the `DataSource` config, network access from the **worker** process to Postgres, and pool sizing. | +| `WorkflowDefinitionException: DB sync is not supported for prefix data attributes` | You used `createByPrefix` with a `DbAttributeSync`; use a fixed key. | + +## Security notes + +- **Credentials** are never held by the SDK — you supply a configured `DataSource`. +- **Values** (primary-key value and the data value) are always bound via `PreparedStatement` parameters. +- **Identifiers** (table, PK column, data column) cannot be bound; they are validated against a strict + allowlist and double-quoted, so a malformed identifier is rejected rather than injected. diff --git a/script/.env b/script/.env index eca4ba6c..c23c1bbb 100644 --- a/script/.env +++ b/script/.env @@ -1,5 +1,5 @@ CASSANDRA_VERSION=3.11.9 -ELASTICSEARCH_VERSION=7.16.2 +ELASTICSEARCH_VERSION=7.17.28 MYSQL_VERSION=8 POSTGRESQL_VERSION=13 TEMPORAL_VERSION=1.25 diff --git a/script/docker-compose.yml b/script/docker-compose.yml index 82d72341..41354912 100644 --- a/script/docker-compose.yml +++ b/script/docker-compose.yml @@ -66,6 +66,20 @@ services: volumes: - ./docker-compose-init.sh:/etc/temporal/init.sh entrypoint: sh -c "/etc/temporal/init.sh" + iwf-integ-postgres: + container_name: iwf-integ-postgres + image: postgres:${POSTGRESQL_VERSION} + environment: + POSTGRES_DB: iwf_integ + POSTGRES_USER: iwf + POSTGRES_PASSWORD: iwf + ports: + # host port 5455 (5432 is often occupied by other local Postgres containers); container is 5432 + - 5455:5432 + volumes: + - ./postgres-init:/docker-entrypoint-initdb.d + networks: + - temporal-network iwf-server: container_name: iwf-server image: iworkflowio/iwf-server:latest #NOTE: you can change to version tag like v1.0.0-RC6 diff --git a/script/postgres-init/iwf-integ-init.sql b/script/postgres-init/iwf-integ-init.sql new file mode 100644 index 00000000..04e56447 --- /dev/null +++ b/script/postgres-init/iwf-integ-init.sql @@ -0,0 +1,14 @@ +-- Initializes the schema/table and the initial dataset used by the data-attribute DB-sync +-- integration test (io.iworkflow.integ.DbSyncTest). Postgres runs any file in +-- /docker-entrypoint-initdb.d/ once on first container start, against POSTGRES_DB (iwf_integ). + +CREATE TABLE IF NOT EXISTS iwf_integ_data ( + pk TEXT PRIMARY KEY, + data_col TEXT +); + +-- The value is stored as JSON text (the SDK mirrors the data attribute's ObjectEncoder JSON), so a +-- String data attribute of value seeded-value is stored as the JSON string "seeded-value". +INSERT INTO iwf_integ_data (pk, data_col) +VALUES ('db-sync-test-key', '"seeded-value"') +ON CONFLICT (pk) DO UPDATE SET data_col = EXCLUDED.data_col; diff --git a/src/main/java/io/iworkflow/core/Client.java b/src/main/java/io/iworkflow/core/Client.java index f923c8c4..9ca4e5f0 100644 --- a/src/main/java/io/iworkflow/core/Client.java +++ b/src/main/java/io/iworkflow/core/Client.java @@ -175,6 +175,14 @@ public String startWorkflow( if (stateOptions != null) { unregisterWorkflowOptions.startStateOptions(stateOptions); } + + // If the workflow has DB-synced data attributes, start at the load system state instead of + // the real starting state. The load state reads the mapped columns and then transitions to + // the real starting state, passing the original input (sent as stateInput) through. + if (!registry.getDbAttributeSyncs(wfType).isEmpty()) { + startStateId = WorkerService.LOAD_DATA_ATTRIBUTES_FROM_DB_STATE_ID; + unregisterWorkflowOptions.startStateOptions(new WorkflowStateOptions().skipWaitUntil(true)); + } } final PersistenceOptions schemaOptions = registry.getPersistenceOptions(wfType); diff --git a/src/main/java/io/iworkflow/core/Registry.java b/src/main/java/io/iworkflow/core/Registry.java index 5a062d96..39493f3a 100644 --- a/src/main/java/io/iworkflow/core/Registry.java +++ b/src/main/java/io/iworkflow/core/Registry.java @@ -3,6 +3,7 @@ import io.iworkflow.core.communication.InternalChannelDef; import io.iworkflow.core.communication.SignalChannelDef; import io.iworkflow.core.persistence.DataAttributeDef; +import io.iworkflow.core.persistence.DbAttributeSync; import io.iworkflow.core.persistence.PersistenceFieldDef; import io.iworkflow.core.persistence.PersistenceOptions; import io.iworkflow.core.persistence.SearchAttributeDef; @@ -10,6 +11,7 @@ import java.lang.reflect.Method; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -27,6 +29,8 @@ public class Registry { private final Map signalTypeStore = new HashMap<>(); private final Map internalChannelTypeStore = new HashMap<>(); private final Map dataAttributeTypeStore = new HashMap<>(); + // (workflow type) -> (data attribute key -> DB sync mapping) + private final Map> dbAttributeSyncStore = new HashMap<>(); private final Map> searchAttributeTypeStore = new HashMap<>(); @@ -153,6 +157,17 @@ private void registerWorkflowDataAttributes(final ObjectWorkflow wf) { for (final DataAttributeDef dataAttributeField : fields) { typeStore.addToStore(dataAttributeField); + + final DbAttributeSync dbSync = dataAttributeField.getDbSync(); + if (dbSync != null) { + if (Boolean.TRUE.equals(dataAttributeField.isPrefix())) { + throw new WorkflowDefinitionException( + "DB sync is not supported for prefix data attributes: " + dataAttributeField.getKey()); + } + dbAttributeSyncStore + .computeIfAbsent(workflowType, s -> new HashMap<>()) + .put(dataAttributeField.getKey(), dbSync); + } } } @@ -250,6 +265,14 @@ public TypeStore getDataAttributeTypeStore(final String workflowType) { return dataAttributeTypeStore.get(workflowType); } + /** + * @return the (data attribute key to DB sync mapping) map for the workflow type, or an empty map + * when the workflow has no DB-synced data attributes. + */ + public Map getDbAttributeSyncs(final String workflowType) { + return dbAttributeSyncStore.getOrDefault(workflowType, Collections.emptyMap()); + } + public Map getSearchAttributeKeyToTypeMap(final String workflowType) { return searchAttributeTypeStore.get(workflowType); } diff --git a/src/main/java/io/iworkflow/core/WorkerService.java b/src/main/java/io/iworkflow/core/WorkerService.java index b5257750..f7e507ba 100644 --- a/src/main/java/io/iworkflow/core/WorkerService.java +++ b/src/main/java/io/iworkflow/core/WorkerService.java @@ -6,7 +6,11 @@ import io.iworkflow.core.mapper.CommandRequestMapper; import io.iworkflow.core.mapper.CommandResultsMapper; import io.iworkflow.core.mapper.StateDecisionMapper; +import io.iworkflow.core.db.DbSyncStatePayload; +import io.iworkflow.core.db.PostgresDataAttributeSyncer; +import io.iworkflow.core.persistence.CellLocation; import io.iworkflow.core.persistence.DataAttributesRWImpl; +import io.iworkflow.core.persistence.DbAttributeSync; import io.iworkflow.core.persistence.Persistence; import io.iworkflow.core.persistence.PersistenceImpl; import io.iworkflow.core.persistence.SearchAttributeRWImpl; @@ -39,6 +43,13 @@ public class WorkerService { public static final String WORKFLOW_WORKER_RPC_API_PATH = "/api/v1/workflowWorker/rpc"; + // Reserved stateIds for the data-attribute DB-sync system states. These are handled specially in + // this worker and are NOT registered in the Registry. They deliberately do not use the server- + // interpreted "_SYS_" prefix, and movements targeting them are always built directly in generated + // form (never routed through StateMovementMapper), so they need no registry validation. + public static final String LOAD_DATA_ATTRIBUTES_FROM_DB_STATE_ID = "_iwf_LoadDataAttributesFromDbState"; + public static final String SYNC_DATA_ATTRIBUTES_TO_DB_STATE_ID = "_iwf_SyncDataAttributesToDbState"; + private final Registry registry; private final WorkerOptions workerOptions; @@ -162,6 +173,14 @@ public WorkflowWorkerRpcResponse handleWorkflowWorkerRpc(final WorkflowWorkerRpc } public WorkflowStateWaitUntilResponse handleWorkflowStateWaitUntil(final WorkflowStateWaitUntilRequest req) { + // The DB-sync system states skip waitUntil (they set skipWaitUntil=true). This is a defensive + // guard: they are not registered, so the registry lookup below would otherwise throw. + if (LOAD_DATA_ATTRIBUTES_FROM_DB_STATE_ID.equals(req.getWorkflowStateId()) + || SYNC_DATA_ATTRIBUTES_TO_DB_STATE_ID.equals(req.getWorkflowStateId())) { + return new WorkflowStateWaitUntilResponse() + .commandRequest(CommandRequestMapper.toGenerated(CommandRequest.empty)); + } + StateDef state = registry.getWorkflowState(req.getWorkflowType(), req.getWorkflowStateId()); final EncodedObject stateInput = req.getStateInput(); final Object input = workerOptions.getObjectEncoder().decode(stateInput, state.getWorkflowState().getInputType()); @@ -228,6 +247,15 @@ public WorkflowStateWaitUntilResponse handleWorkflowStateWaitUntil(final Workflo } public WorkflowStateExecuteResponse handleWorkflowStateExecute(final WorkflowStateExecuteRequest req) { + // The DB-sync system states are handled specially and are not registered in the Registry, so + // they must be intercepted before the registry lookup below (which would otherwise throw). + if (LOAD_DATA_ATTRIBUTES_FROM_DB_STATE_ID.equals(req.getWorkflowStateId())) { + return handleLoadDataAttributesFromDb(req); + } + if (SYNC_DATA_ATTRIBUTES_TO_DB_STATE_ID.equals(req.getWorkflowStateId())) { + return handleSyncDataAttributesToDb(req); + } + StateDef state = registry.getWorkflowState(req.getWorkflowType(), req.getWorkflowStateId()); final Object input; final EncodedObject stateInput = req.getStateInput(); @@ -265,11 +293,36 @@ public WorkflowStateExecuteResponse handleWorkflowStateExecute(final WorkflowSta throw new InvalidStateDecisionException("State decision returned by execute method cannot be null or empty"); } + final List toReturnToServer = dataObjectsRW.getToReturnToServer(); + + // If this state mutated any DB-synced data attribute, reroute the decision through the sync + // system state (carrying the captured original decision), so the DB write happens after the + // iWF server has persisted the mutations. See DbSyncStatePayload. + final Map dbSyncs = registry.getDbAttributeSyncs(req.getWorkflowType()); + final List mutatedMappedKeys = toReturnToServer.stream() + .map(KeyValue::getKey) + .filter(dbSyncs::containsKey) + .collect(Collectors.toList()); + + final io.iworkflow.gen.models.StateDecision generatedDecision; + if (mutatedMappedKeys.isEmpty()) { + generatedDecision = StateDecisionMapper.toGenerated(stateDecision, req.getWorkflowType(), registry, workerOptions.getObjectEncoder()); + } else { + final io.iworkflow.gen.models.StateDecision original = + StateDecisionMapper.toGenerated(stateDecision, req.getWorkflowType(), registry, workerOptions.getObjectEncoder()); + final DbSyncStatePayload payload = new DbSyncStatePayload(original, mutatedMappedKeys); + generatedDecision = new io.iworkflow.gen.models.StateDecision() + .addNextStatesItem(new io.iworkflow.gen.models.StateMovement() + .stateId(SYNC_DATA_ATTRIBUTES_TO_DB_STATE_ID) + .stateInput(workerOptions.getObjectEncoder().encode(payload)) + .stateOptions(new io.iworkflow.gen.models.WorkflowStateOptions().skipWaitUntil(true))); + } + final WorkflowStateExecuteResponse response = new WorkflowStateExecuteResponse() - .stateDecision(StateDecisionMapper.toGenerated(stateDecision, req.getWorkflowType(), registry, workerOptions.getObjectEncoder())); + .stateDecision(generatedDecision); - if (dataObjectsRW.getToReturnToServer().size() > 0) { - response.upsertDataObjects(dataObjectsRW.getToReturnToServer()); + if (toReturnToServer.size() > 0) { + response.upsertDataObjects(toReturnToServer); } if (stateExeLocals.getUpsertStateExecutionLocalAttributes().size() > 0) { response.upsertStateLocals(stateExeLocals.getUpsertStateExecutionLocalAttributes()); @@ -296,6 +349,91 @@ public WorkflowStateExecuteResponse handleWorkflowStateExecute(final WorkflowSta return response; } + /** + * Handles the (unregistered, worker-only) load system state that runs as the workflow's first state + * when the workflow has DB-synced data attributes. It loads each mapped data attribute from its + * Postgres cell, then transitions to the real starting state, passing the original start input through. + */ + private WorkflowStateExecuteResponse handleLoadDataAttributesFromDb(final WorkflowStateExecuteRequest req) { + final String workflowType = req.getWorkflowType(); + final ObjectEncoder encoder = workerOptions.getObjectEncoder(); + + final DataAttributesRWImpl dataObjectsRW = createDataObjectsRW(workflowType, req.getDataObjects()); + final Context context = fromIdlContext(req.getContext(), workflowType); + final Map saTypeMap = registry.getSearchAttributeKeyToTypeMap(workflowType); + final SearchAttributeRWImpl searchAttributeRW = new SearchAttributeRWImpl(saTypeMap, req.getSearchAttributes()); + final StateExecutionLocalsImpl stateExeLocals = new StateExecutionLocalsImpl(toMap(null), encoder); + final Persistence persistence = new PersistenceImpl(dataObjectsRW, searchAttributeRW, stateExeLocals); + + final TypeStore typeStore = registry.getDataAttributeTypeStore(workflowType); + final Map dbSyncs = registry.getDbAttributeSyncs(workflowType); + for (final Map.Entry entry : dbSyncs.entrySet()) { + final String key = entry.getKey(); + final DbAttributeSync sync = entry.getValue(); + final CellLocation loc = sync.getLocator().apply(context.getWorkflowId(), persistence); + final String columnText = PostgresDataAttributeSyncer.select( + sync.getDataSource(), sync.getTableName(), sync.getPkColumnName(), loc.getColumnName(), loc.getPkValue()); + if (columnText != null) { + final Class registeredType = typeStore.getType(key); + final Object value = encoder.decode( + new EncodedObject().encoding(encoder.getEncodingType()).data(columnText), registeredType); + persistence.setDataAttribute(key, value); + } + } + + final Optional startStateOptional = registry.getWorkflowStartingState(workflowType); + if (!startStateOptional.isPresent()) { + throw new WorkflowDefinitionException( + "the DB-sync load state requires a starting state in workflow " + workflowType); + } + final WorkflowState realStartState = startStateOptional.get().getWorkflowState(); + final Object originalInput = encoder.decode(req.getStateInput(), realStartState.getInputType()); + final StateDecision decision = StateDecision.singleNextState(realStartState.getStateId(), originalInput, null); + + final WorkflowStateExecuteResponse response = new WorkflowStateExecuteResponse() + .stateDecision(StateDecisionMapper.toGenerated(decision, workflowType, registry, encoder)); + final List toReturnToServer = dataObjectsRW.getToReturnToServer(); + if (toReturnToServer.size() > 0) { + response.upsertDataObjects(toReturnToServer); + } + return response; + } + + /** + * Handles the (unregistered, worker-only) sync system state. It writes each carried data attribute's + * current value to its Postgres cell (the value has already been persisted by the iWF server), then + * replays the captured original decision verbatim so the workflow proceeds as intended. + */ + private WorkflowStateExecuteResponse handleSyncDataAttributesToDb(final WorkflowStateExecuteRequest req) { + final String workflowType = req.getWorkflowType(); + final ObjectEncoder encoder = workerOptions.getObjectEncoder(); + final DbSyncStatePayload payload = encoder.decode(req.getStateInput(), DbSyncStatePayload.class); + + final DataAttributesRWImpl dataObjectsRW = createDataObjectsRW(workflowType, req.getDataObjects()); + final Context context = fromIdlContext(req.getContext(), workflowType); + final Map saTypeMap = registry.getSearchAttributeKeyToTypeMap(workflowType); + final SearchAttributeRWImpl searchAttributeRW = new SearchAttributeRWImpl(saTypeMap, req.getSearchAttributes()); + final StateExecutionLocalsImpl stateExeLocals = new StateExecutionLocalsImpl(toMap(null), encoder); + final Persistence persistence = new PersistenceImpl(dataObjectsRW, searchAttributeRW, stateExeLocals); + + final Map currentValues = toMap(req.getDataObjects()); + final Map dbSyncs = registry.getDbAttributeSyncs(workflowType); + for (final String key : payload.getDataAttributeKeys()) { + final DbAttributeSync sync = dbSyncs.get(key); + if (sync == null) { + continue; + } + final CellLocation loc = sync.getLocator().apply(context.getWorkflowId(), persistence); + final EncodedObject current = currentValues.get(key); + final String columnText = current == null ? null : current.getData(); + PostgresDataAttributeSyncer.update( + sync.getDataSource(), sync.getTableName(), sync.getPkColumnName(), + loc.getColumnName(), loc.getPkValue(), columnText); + } + + return new WorkflowStateExecuteResponse().stateDecision(payload.getOriginalDecision()); + } + private List toInterStateChannelPublishing(final Map> toPublish) { List results = new ArrayList<>(); toPublish.forEach((cname, list) -> { diff --git a/src/main/java/io/iworkflow/core/db/DbSyncStatePayload.java b/src/main/java/io/iworkflow/core/db/DbSyncStatePayload.java new file mode 100644 index 00000000..5f1c0889 --- /dev/null +++ b/src/main/java/io/iworkflow/core/db/DbSyncStatePayload.java @@ -0,0 +1,45 @@ +package io.iworkflow.core.db; + +import io.iworkflow.gen.models.StateDecision; + +import java.util.List; + +/** + * The input carried into the (unregistered, worker-only) SyncDataAttributesToDb system state. + *

+ * When a normal state's execute method mutates a DB-mapped data attribute, the worker reroutes that + * state's decision through the sync state. To let the workflow continue exactly as intended, the + * original decision is captured in its already-generated form ({@link StateDecision}, which is + * Jackson-serializable) and carried here, together with the list of mutated data-attribute keys that + * must be written to the database. The sync state performs the writes and then replays + * {@link #getOriginalDecision()} verbatim. + */ +public class DbSyncStatePayload { + private StateDecision originalDecision; + private List dataAttributeKeys; + + // for Jackson deserialization + public DbSyncStatePayload() { + } + + public DbSyncStatePayload(final StateDecision originalDecision, final List dataAttributeKeys) { + this.originalDecision = originalDecision; + this.dataAttributeKeys = dataAttributeKeys; + } + + public StateDecision getOriginalDecision() { + return originalDecision; + } + + public void setOriginalDecision(final StateDecision originalDecision) { + this.originalDecision = originalDecision; + } + + public List getDataAttributeKeys() { + return dataAttributeKeys; + } + + public void setDataAttributeKeys(final List dataAttributeKeys) { + this.dataAttributeKeys = dataAttributeKeys; + } +} diff --git a/src/main/java/io/iworkflow/core/db/PostgresDataAttributeSyncer.java b/src/main/java/io/iworkflow/core/db/PostgresDataAttributeSyncer.java new file mode 100644 index 00000000..2bad75b3 --- /dev/null +++ b/src/main/java/io/iworkflow/core/db/PostgresDataAttributeSyncer.java @@ -0,0 +1,121 @@ +package io.iworkflow.core.db; + +import javax.sql.DataSource; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.regex.Pattern; + +/** + * Executes the SELECT (load) and UPDATE (sync) against a user-owned Postgres table for the + * data-attribute-to-column mirroring feature. + *

+ * Security: + *

    + *
  • All values (primary key value, data value) are bound via {@link PreparedStatement} + * parameters — never string-concatenated.
    • + *
  • All identifiers (table, primary-key column, data column) come from developer config, + * but they cannot be bound as parameters. They are validated against a strict allowlist and + * double-quoted, so a malformed/malicious identifier is rejected rather than injected.
    • + *
  • Connections are obtained from the caller-supplied {@link DataSource} and always used in a + * try-with-resources block to avoid connection leaks.
    • + *
+ */ +public final class PostgresDataAttributeSyncer { + + // A safe SQL identifier: starts with a letter/underscore, followed by letters/digits/underscores. + private static final Pattern SAFE_IDENTIFIER = Pattern.compile("^[A-Za-z_][A-Za-z0-9_]*$"); + + private PostgresDataAttributeSyncer() { + } + + /** + * @return the raw text stored in the cell, or null if the row does not exist or the column is null. + */ + public static String select( + final DataSource dataSource, + final String tableName, + final String pkColumnName, + final String dataColumnName, + final String pkValue) { + final String sql = String.format( + "SELECT %s FROM %s WHERE %s = ?", + quoteIdentifier(dataColumnName), + quoteQualifiedIdentifier(tableName), + quoteIdentifier(pkColumnName)); + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, pkValue); + try (ResultSet resultSet = statement.executeQuery()) { + if (resultSet.next()) { + return resultSet.getString(1); + } + return null; + } + } catch (SQLException e) { + throw new RuntimeException( + String.format("failed to load data attribute from %s.%s where %s=%s", + tableName, dataColumnName, pkColumnName, pkValue), + e); + } + } + + /** + * Updates the cell with the given text value. This is an idempotent single-row UPDATE, so it is + * safe to retry (e.g. when the sync state is retried by the iWF server). + */ + public static void update( + final DataSource dataSource, + final String tableName, + final String pkColumnName, + final String dataColumnName, + final String pkValue, + final String value) { + final String sql = String.format( + "UPDATE %s SET %s = ? WHERE %s = ?", + quoteQualifiedIdentifier(tableName), + quoteIdentifier(dataColumnName), + quoteIdentifier(pkColumnName)); + try (Connection connection = dataSource.getConnection(); + PreparedStatement statement = connection.prepareStatement(sql)) { + statement.setString(1, value); + statement.setString(2, pkValue); + statement.executeUpdate(); + } catch (SQLException e) { + throw new RuntimeException( + String.format("failed to sync data attribute to %s.%s where %s=%s", + tableName, dataColumnName, pkColumnName, pkValue), + e); + } + } + + /** + * Validates a single SQL identifier against the allowlist and wraps it in double quotes. + */ + static String quoteIdentifier(final String identifier) { + if (identifier == null || !SAFE_IDENTIFIER.matcher(identifier).matches()) { + throw new IllegalArgumentException("unsafe SQL identifier: " + identifier); + } + return "\"" + identifier + "\""; + } + + /** + * Validates and quotes a possibly schema-qualified identifier (e.g. "public.orders"), + * quoting each dot-separated part independently. + */ + static String quoteQualifiedIdentifier(final String identifier) { + if (identifier == null || identifier.isEmpty()) { + throw new IllegalArgumentException("unsafe SQL identifier: " + identifier); + } + final String[] parts = identifier.split("\\.", -1); + final StringBuilder sb = new StringBuilder(); + for (int i = 0; i < parts.length; i++) { + if (i > 0) { + sb.append('.'); + } + sb.append(quoteIdentifier(parts[i])); + } + return sb.toString(); + } +} diff --git a/src/main/java/io/iworkflow/core/persistence/CellLocation.java b/src/main/java/io/iworkflow/core/persistence/CellLocation.java new file mode 100644 index 00000000..df37796f --- /dev/null +++ b/src/main/java/io/iworkflow/core/persistence/CellLocation.java @@ -0,0 +1,32 @@ +package io.iworkflow.core.persistence; + +/** + * Identifies a single cell (row + column) in a user-owned database table that a data attribute is + * mirrored to. It is the return value of the locator lambda in {@link DbAttributeSync}. + *

+ * The primary key value is a String — it is bound as a {@link java.sql.PreparedStatement} parameter, + * so it can represent any text/UUID/numeric primary key as long as the column type is compatible. + */ +public final class CellLocation { + private final String pkValue; + private final String columnName; + + public CellLocation(final String pkValue, final String columnName) { + if (pkValue == null) { + throw new IllegalArgumentException("pkValue cannot be null"); + } + if (columnName == null) { + throw new IllegalArgumentException("columnName cannot be null"); + } + this.pkValue = pkValue; + this.columnName = columnName; + } + + public String getPkValue() { + return pkValue; + } + + public String getColumnName() { + return columnName; + } +} diff --git a/src/main/java/io/iworkflow/core/persistence/DataAttributeDef.java b/src/main/java/io/iworkflow/core/persistence/DataAttributeDef.java index 18999830..70391b1f 100644 --- a/src/main/java/io/iworkflow/core/persistence/DataAttributeDef.java +++ b/src/main/java/io/iworkflow/core/persistence/DataAttributeDef.java @@ -2,11 +2,21 @@ import org.immutables.value.Value; +import javax.annotation.Nullable; + @Value.Immutable public abstract class DataAttributeDef implements PersistenceFieldDef { public abstract Class getDataAttributeType(); public abstract Boolean isPrefix(); + /** + * @return optional configuration to automatically mirror this data attribute to a cell of a + * user-owned Postgres table (load on workflow start, sync on execute-mutation). Null when the + * data attribute is not DB-synced. Not supported together with prefix data attributes. + */ + @Nullable + public abstract DbAttributeSync getDbSync(); + /** * iWF will verify if the key has been registered for the data attribute created using this method, * allowing users to create only one data attribute with the same key and data type. @@ -23,6 +33,26 @@ public static DataAttributeDef create(final Class dataType, final String key) { .build(); } + /** + * Creates a data attribute that is automatically mirrored to a cell of a user-owned Postgres table: + * the value is loaded from the cell when the workflow starts, and written back to the cell whenever + * the data attribute is mutated in a state's execute method. The iWF data attribute remains the + * source of truth; the database cell is a mirror. + * + * @param dataType required. + * @param key required. The unique key. + * @param dbSync required. The table/column mapping and DataSource. See {@link DbAttributeSync}. + * @return a data attribute definition with DB sync enabled + */ + public static DataAttributeDef create(final Class dataType, final String key, final DbAttributeSync dbSync) { + return ImmutableDataAttributeDef.builder() + .key(key) + .dataAttributeType(dataType) + .isPrefix(false) + .dbSync(dbSync) + .build(); + } + /** * iWF now supports dynamically created data attributes with a shared prefix and the same data type. * (E.g., dynamically created data attributes of type String can be named with a common prefix like: data_attribute_prefix_1: "one", data_attribute_prefix_2: "two") diff --git a/src/main/java/io/iworkflow/core/persistence/DbAttributeSync.java b/src/main/java/io/iworkflow/core/persistence/DbAttributeSync.java new file mode 100644 index 00000000..af71d566 --- /dev/null +++ b/src/main/java/io/iworkflow/core/persistence/DbAttributeSync.java @@ -0,0 +1,81 @@ +package io.iworkflow.core.persistence; + +import javax.sql.DataSource; +import java.util.function.BiFunction; + +/** + * Configuration that maps a single data attribute to a cell (row + column) of a user-owned Postgres + * table so that the data attribute is: + *

    + *
  • loaded from the cell when the workflow starts, and
    • + *
  • written back to the cell whenever the data attribute is mutated in a state's execute method.
    • + *
+ * The iWF data attribute remains the source of truth; the database cell is a mirror. + *

+ * The SDK never stores database credentials — the caller supplies a fully-configured + * {@link javax.sql.DataSource} (owning its own connection pooling and credentials). + * + * @see DataAttributeDef#create(Class, String, DbAttributeSync) + */ +public final class DbAttributeSync { + + private final DataSource dataSource; + private final String tableName; + private final String pkColumnName; + private final BiFunction locator; + + private DbAttributeSync( + final DataSource dataSource, + final String tableName, + final String pkColumnName, + final BiFunction locator) { + if (dataSource == null) { + throw new IllegalArgumentException("dataSource cannot be null"); + } + if (tableName == null) { + throw new IllegalArgumentException("tableName cannot be null"); + } + if (pkColumnName == null) { + throw new IllegalArgumentException("pkColumnName cannot be null"); + } + if (locator == null) { + throw new IllegalArgumentException("locator cannot be null"); + } + this.dataSource = dataSource; + this.tableName = tableName; + this.pkColumnName = pkColumnName; + this.locator = locator; + } + + /** + * @param dataSource required. A user-owned, fully-configured DataSource (credentials + pooling). + * @param tableName required. The table name, optionally schema-qualified (e.g. "public.orders"). + * @param pkColumnName required. The primary-key column name used in the WHERE clause. + * @param locator required. Given the workflowId and the current {@link Persistence}, returns the + * {@link CellLocation} (primary-key value + data column name) to load from / sync to. + * @return a DbAttributeSync configuration + */ + public static DbAttributeSync of( + final DataSource dataSource, + final String tableName, + final String pkColumnName, + final BiFunction locator) { + return new DbAttributeSync(dataSource, tableName, pkColumnName, locator); + } + + public DataSource getDataSource() { + return dataSource; + } + + public String getTableName() { + return tableName; + } + + public String getPkColumnName() { + return pkColumnName; + } + + public BiFunction getLocator() { + return locator; + } +} diff --git a/src/test/java/io/iworkflow/integ/DbSyncTest.java b/src/test/java/io/iworkflow/integ/DbSyncTest.java new file mode 100644 index 00000000..bb832146 --- /dev/null +++ b/src/test/java/io/iworkflow/integ/DbSyncTest.java @@ -0,0 +1,98 @@ +package io.iworkflow.integ; + +import io.iworkflow.core.Client; +import io.iworkflow.core.ClientOptions; +import io.iworkflow.core.WorkflowOptions; +import io.iworkflow.gen.models.IDReusePolicy; +import io.iworkflow.integ.dbsync.DbSyncWorkflow; +import io.iworkflow.integ.dbsync.DbSyncWorkflowState1; +import io.iworkflow.spring.TestSingletonWorkerService; +import io.iworkflow.spring.controller.WorkflowRegistry; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.util.concurrent.ExecutionException; + +import javax.sql.DataSource; + +/** + * End-to-end test of the data-attribute Postgres DB-sync feature against a real Postgres + * (script/docker-compose.yml: iwf-integ-postgres) and a live iWF server. + *

+ * The schema, table, and initial dataset are created by script/postgres-init/iwf-integ-init.sql on + * first container start. @BeforeEach resets the target row to the known initial value so the test is + * deterministic across reruns against a persistent local DB. + */ +public class DbSyncTest { + + private static final String INITIAL_DB_JSON = "\"seeded-value\""; + private static final String INITIAL_DB_VALUE = "seeded-value"; + + private final DataSource dataSource = DbSyncWorkflow.buildDataSource(); + + @BeforeEach + public void setup() throws ExecutionException, InterruptedException { + TestSingletonWorkerService.startWorkerIfNotUp(); + resetSeedRow(); + } + + private void resetSeedRow() { + final String sql = "INSERT INTO " + DbSyncWorkflow.TABLE_NAME + + " (pk, data_col) VALUES (?, ?) ON CONFLICT (pk) DO UPDATE SET data_col = EXCLUDED.data_col"; + try (Connection c = dataSource.getConnection(); + PreparedStatement ps = c.prepareStatement(sql)) { + ps.setString(1, DbSyncWorkflow.ROW_PK); + ps.setString(2, INITIAL_DB_JSON); + ps.executeUpdate(); + } catch (Exception e) { + throw new RuntimeException("failed to reset seed row (is iwf-integ-postgres up?)", e); + } + } + + private String readColumn() { + final String sql = "SELECT data_col FROM " + DbSyncWorkflow.TABLE_NAME + " WHERE pk = ?"; + try (Connection c = dataSource.getConnection(); + PreparedStatement ps = c.prepareStatement(sql)) { + ps.setString(1, DbSyncWorkflow.ROW_PK); + try (ResultSet rs = ps.executeQuery()) { + return rs.next() ? rs.getString(1) : null; + } + } catch (Exception e) { + throw new RuntimeException("failed to read column", e); + } + } + + // In CI (Linux) the iWF server reaches the host worker via 172.17.0.1 (ClientOptions.localDefault). + // On macOS/OrbStack the host is instead reachable via host.docker.internal, so set + // IWF_WORKER_URL=http://host.docker.internal:8802 when running locally there. + private static ClientOptions clientOptions() { + final String workerUrl = System.getenv("IWF_WORKER_URL"); + if (workerUrl != null && !workerUrl.isEmpty()) { + return ClientOptions.minimum(workerUrl, ClientOptions.defaultServerUrl); + } + return ClientOptions.localDefault; + } + + @Test + public void testDataAttributeDbSync() { + final Client client = new Client(WorkflowRegistry.registry, clientOptions()); + final String wfId = "db-sync-test-id-" + System.currentTimeMillis() / 1000; + + client.startWorkflow( + DbSyncWorkflow.class, wfId, 10, "start", + WorkflowOptions.basicBuilder() + .workflowIdReusePolicy(IDReusePolicy.ALLOW_IF_NO_RUNNING) + .build()); + + // (a) load-on-start: the state read the seeded DB value, returned as the workflow output + final String output = client.getSimpleWorkflowResultWithWait(String.class, wfId); + Assertions.assertEquals(INITIAL_DB_VALUE, output); + + // (b) sync-on-mutation: the mutated value was written back to the DB cell (stored as JSON text) + Assertions.assertEquals("\"" + DbSyncWorkflowState1.MUTATED_VALUE + "\"", readColumn()); + } +} diff --git a/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflow.java b/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflow.java new file mode 100644 index 00000000..138ed469 --- /dev/null +++ b/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflow.java @@ -0,0 +1,62 @@ +package io.iworkflow.integ.dbsync; + +import io.iworkflow.core.ObjectWorkflow; +import io.iworkflow.core.StateDef; +import io.iworkflow.core.persistence.CellLocation; +import io.iworkflow.core.persistence.DataAttributeDef; +import io.iworkflow.core.persistence.DbAttributeSync; +import io.iworkflow.core.persistence.PersistenceFieldDef; +import org.postgresql.ds.PGSimpleDataSource; +import org.springframework.stereotype.Component; + +import javax.sql.DataSource; +import java.util.Arrays; +import java.util.List; + +/** + * Integration-test workflow demonstrating the data-attribute Postgres DB-sync feature. + * The "status" data attribute is mirrored to the iwf_integ_data.data_col cell keyed by + * {@link #ROW_PK}. On start it is loaded from that cell; when the state mutates it, it is synced back. + */ +@Component +public class DbSyncWorkflow implements ObjectWorkflow { + + public static final String STATUS_KEY = "status"; + public static final String TABLE_NAME = "iwf_integ_data"; + public static final String PK_COLUMN = "pk"; + public static final String DATA_COLUMN = "data_col"; + public static final String ROW_PK = "db-sync-test-key"; + + public static final String JDBC_URL = "jdbc:postgresql://localhost:5455/iwf_integ"; + public static final String DB_USER = "iwf"; + public static final String DB_PASSWORD = "iwf"; + + public static DataSource buildDataSource() { + final PGSimpleDataSource ds = new PGSimpleDataSource(); + ds.setUrl(JDBC_URL); + ds.setUser(DB_USER); + ds.setPassword(DB_PASSWORD); + return ds; + } + + private static final DataSource DATA_SOURCE = buildDataSource(); + + @Override + public List getWorkflowStates() { + return Arrays.asList(StateDef.startingState(new DbSyncWorkflowState1())); + } + + @Override + public List getPersistenceSchema() { + return Arrays.asList( + DataAttributeDef.create( + String.class, + STATUS_KEY, + DbAttributeSync.of( + DATA_SOURCE, + TABLE_NAME, + PK_COLUMN, + (workflowId, persistence) -> new CellLocation(ROW_PK, DATA_COLUMN))) + ); + } +} diff --git a/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflowState1.java b/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflowState1.java new file mode 100644 index 00000000..55c4b4dc --- /dev/null +++ b/src/test/java/io/iworkflow/integ/dbsync/DbSyncWorkflowState1.java @@ -0,0 +1,37 @@ +package io.iworkflow.integ.dbsync; + +import io.iworkflow.core.Context; +import io.iworkflow.core.StateDecision; +import io.iworkflow.core.WorkflowState; +import io.iworkflow.core.command.CommandResults; +import io.iworkflow.core.communication.Communication; +import io.iworkflow.core.persistence.Persistence; + +public class DbSyncWorkflowState1 implements WorkflowState { + + // the value the state writes to the data attribute; this is what should end up mirrored to the DB + public static final String MUTATED_VALUE = "synced-value"; + + @Override + public Class getInputType() { + return String.class; + } + + @Override + public StateDecision execute( + final Context context, + final String input, + final CommandResults commandResults, + final Persistence persistence, + final Communication communication) { + // the value loaded from the DB cell at workflow start (proves load-on-start) + final String loadedFromDb = persistence.getDataAttribute(DbSyncWorkflow.STATUS_KEY, String.class); + + // mutate the DB-synced data attribute (proves sync-on-mutation): after this state, the worker + // reroutes the decision through the sync state which writes MUTATED_VALUE to the DB cell. + persistence.setDataAttribute(DbSyncWorkflow.STATUS_KEY, MUTATED_VALUE); + + // complete the workflow, returning the loaded value so the test can assert load worked + return StateDecision.gracefulCompleteWorkflow(loadedFromDb); + } +}