diff --git a/.circleci/config.yml b/.circleci/config.yml index e41c5c7..02caf7a 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -21,6 +21,13 @@ jobs: npm test npm run test:integration npm run build + deployment-checks: + docker: + - image: cimg/python:3.13.2 + steps: + - checkout + - run: pip install boto3 + - run: python3 -m unittest discover -s deploy/tests deploy: docker: - image: cimg/python:3.13.2-browsers @@ -46,7 +53,7 @@ jobs: name: Build tested runtime image command: docker build --tag forms-processor-v6:candidate . - run: - name: Deploy through CloudFormation + name: Deploy a new task revision to the existing ECS service no_output_timeout: 35m command: | git clone --quiet --depth 1 --branch v1.4.20 https://github.com/topcoder-platform/tc-deploy-scripts ../buildscript @@ -64,10 +71,11 @@ workflows: build-and-deploy: jobs: - quality + - deployment-checks - deploy: name: deploy-dev context: org-global - requires: [quality] + requires: [quality, deployment-checks] deploy_env: DEV environment: dev serial-group: << pipeline.project.slug >>/deploy-dev @@ -77,7 +85,7 @@ workflows: - deploy: name: deploy-production context: org-global - requires: [quality] + requires: [quality, deployment-checks] deploy_env: PROD environment: production serial-group: << pipeline.project.slug >>/deploy-production diff --git a/.env.example b/.env.example index 8b21c97..b214261 100644 --- a/.env.example +++ b/.env.example @@ -12,3 +12,8 @@ KAFKA_FROM_BEGINNING=true # AUTH0_CLIENT_SECRET= # AUTH0_PROXY_SERVER_URL= PORT=3000 + +# Authoritative settings for lets-talk-sales-email; restart after changes. +# Empty values defer delivery even if stale database settings exist. +LETS_TALK_RECIPIENTS= +LETS_TALK_SENDGRID_TEMPLATE_ID= diff --git a/README.md b/README.md index a3d8110..dff6472 100644 --- a/README.md +++ b/README.md @@ -3,12 +3,12 @@ A small TypeScript Kafka worker for the HubSpot forms replacement. Consumes the `forms-api-v6` Bus API envelope on `form.submitted`, saves a durable inbox record, then routes each submission to independently retried actions. The first action -publishes `lets-talk` email requests through Bus API to `external.action.email`; +publishes `lets_talk` email requests through Bus API to `external.action.email`; `email-service-v6` delivers them using the configured SendGrid dynamic template. The initial `lets-talk-sales-email` flow is **disabled**. Recipients and template ID -are TBD (empty/null in the seed). The optional sender is also unset; email-service-v6 -uses its configured default when no sender override is supplied. Matching submissions queue +come from `LETS_TALK_RECIPIENTS` and `LETS_TALK_SENDGRID_TEMPLATE_ID`, injected +from processor SSM SecureStrings. email-service-v6 uses its configured default sender. Matching submissions queue until you configure and enable the flow. No email is sent by a disabled flow. The form integration must include `"kafka": true` in the forms API submission @@ -53,7 +53,8 @@ TEST_DATABASE_URL=postgresql://postgres:forms_local@localhost:5547/forms_process The integration suite replaces its `forms` schema. It never uses DATABASE_URL. `test/fixtures/processor-schema.sql` is a test-only copy of the API migration -`20260928020000_processor_flows`; keep them identical when changing the schema. +`20260928020000_processor_flows`; `test/fixtures/lets-talk-key.sql` copies the +additive key correction. Keep each fixture identical to its API migration. ## Configure the first flow @@ -61,40 +62,54 @@ The canonical Prisma models/migration live in **forms-api-v6**. Deploy migration `20260928020000_processor_flows` before deploying the worker; the worker fails startup if tables are missing. It does not run migrations or modify form schemas. -Run a parameterized update using your database administration tooling: +The website publishes the key `lets_talk`. Apply the follow-up migration +`20261013010000_lets_talk_processor_key` to correct the original `lets-talk` seed; +it preserves configuration and successful receipts, and reroutes retained events +that missed this flow. Do not edit the already-applied original migration. + +Create these SecureStrings and redeploy the processor: + +- `/config/forms-processor-v6/appvar/LETS_TALK_RECIPIENTS`: comma-separated emails. + Whitespace is trimmed and case-insensitive duplicates are removed; empty entries + within a nonempty list are rejected. Up to 1000 addresses are supported. +- `/config/forms-processor-v6/appvar/LETS_TALK_SENDGRID_TEMPLATE_ID`: an active + SendGrid dynamic template ID (`d-` plus 32 lowercase hexadecimal characters). + +These settings replace database settings **only** for `lets-talk-sales-email`. +Missing values defer delivery even when the database contains old valid settings. +Malformed nonempty values fail startup. Parameter changes require redeployment; +other flows still read their database settings on each attempt and may override +`fromEmail`/`fromName`. This sales flow uses email-service-v6's default sender. + +After reviewing pending deliveries and verifying the deployed settings, enable: ```sql -UPDATE forms."ProcessorFlow" -SET settings = jsonb_build_object( - 'recipients', jsonb_build_array('sales@example.com', 'associate@example.com'), - 'templateId', 'd-00000000000000000000000000000000', - 'fromEmail', 'verified-sender@example.com', - 'fromName', 'Topcoder' - ), "updatedAt" = now() -WHERE id = 'lets-talk-sales-email'; --- Replace all sample values before enabling: UPDATE forms."ProcessorFlow" SET enabled = true, "updatedAt" = now() -WHERE id = 'lets-talk-sales-email'; +WHERE id = 'lets-talk-sales-email' AND "formKey" = 'lets_talk' + AND action = 'sendgrid-email'; ``` -Use an active SendGrid dynamic template. A sender override must be verified; omit -`fromEmail` or set it to null to use email-service-v6's configured sender. Changes are read -for each delivery attempt; no restart is needed for recipients/template/sender. The full recipient list is visible in the email's To field. Recipients always -come from trusted flow settings, never the submission. The template defines the +come from trusted configuration, never the submission. The template defines the subject and content; use escaped Handlebars expressions for submitted answers. The published payload is `{ recipients, sendgrid_template_id, version: "v3", data, from? }` inside the standard Bus API envelope. `data` holds the template variables; `from` is emitted only when a sender override is configured. The existing stored action name `sendgrid-email` is retained for compatibility, but always uses Bus API. -No migration or settings rewrite is needed for existing flows or pending receipts. +The form-key correction is additive; successful pending-flow receipts are not replayed. Template data includes `submissionId`, `formKey`, `version`, `submittedAt`, `memberId`, `sourcePage`, `answers`, and `fields` (array of `{key,value}` strings). For lets-talk, use `{{answers.firstname}}`, `{{answers.lastname}}`, `{{answers.email}}`, `{{answers.company}}`, `{{answers.interested_in}}`, and `{{answers.briefly_describe_your_inquiry}}`. `{{#each fields}}` can render arbitrary -forms. Submitted data is nested and cannot overwrite event metadata. +forms. Submitted data is nested and cannot overwrite event metadata. The complete +[example email envelope](examples/lets-talk-email.json) includes every field of the +production form. Its `payload.data` becomes SendGrid `dynamic_template_data` directly: +use `{{answers.firstname}}`, not `{{data.answers.firstname}}`. Select answers contain +stable keys (`crowdsourcing`, `app_design_development`, `freelancers`), not display labels. +The optional inquiry answer may be omitted when not supplied. Use escaped `{{...}}` +for all submitted values; the template owns its subject and content. ## More forms and content-based flows @@ -170,4 +185,5 @@ forms API database login, so a dedicated restricted login is an optional later s See [deploy/README.md](deploy/README.md). CircleCI checks every branch and deploys `develop` → dev, `master` → production using `org-global` credentials and the same pinned Topcoder AWS credential helper as forms-api-v6. Each environment needs its -one-time CloudFormation bootstrap and API migration before its first deployment. +existing ECS service/ECR repository, required SSM references and API migrations before deployment. +The pipeline registers a new task revision and updates the existing service; it does not apply CloudFormation. diff --git a/deploy/README.md b/deploy/README.md index 353c728..e45cda0 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -1,62 +1,93 @@ -# Deployment - -`service.yaml` owns ECR, CloudWatch, execution/task roles, a private security group, -database/Kafka ingress rules, task definition, and the `forms-processor-v6` Fargate -service in `topcoder-infrastructure`. There is no ALB, API Gateway, or public ingress. -The default task is 0.25 vCPU / 512 MiB with a read-only filesystem and graceful stop. - -1. Apply the forms-api-v6 `20260928020000_processor_flows` migration using that API's - normal migration deployment. It adds three tables and a disabled lets-talk flow. -2. Ensure `form.submitted` exists on MSK before starting the service. Dev uses two - partitions and replication factor two. The worker disables topic auto-creation. - Provision the topic using your Kafka administration tooling inside the VPC; - retain the broker's normal retention policy. -3. Bootstrap the CloudFormation stack in each account at DesiredCount=0, passing - its VPC, private subnets, database SG, and MSK SG. Dev values are checked in as - `parameters-dev.json`; supply production's own values, never reuse dev IDs. -4. Verify these existing SSM values are present: forms API DATABASE_URL, shared - KAFKA_URL, shared AUTH0_URL/AUTH0_AUDIENCE/AUTH0_PROXY_SERVER_URL, and the - forms API AUTH0_CLIENT_ID/AUTH0_CLIENT_SECRET. `BusCredentialsPrefix` defaults - to `/config/forms-api-v6/appvar` to reuse its Bus-authorized M2M identity; point - it to processor-specific credentials when provisioned. That client needs - `write:bus_api`. The task receives the environment's `/v6` Bus API base. - The execution role permits only the named SSM values and no SendGrid key. - Set SecretsKmsKeyArn if these parameters use a customer-managed KMS key. -5. Build and release. The script pushes an immutable image tag and updates the - checked-in CloudFormation template while preserving existing stack parameters. - It waits for CloudFormation stabilization/circuit-breaker rollback and verifies - the retained ECS task definition. Startup checks ensure migration tables exist. +# Deployment to existing ECS services -```sh -aws cloudformation create-stack --stack-name forms-processor-v6-dev \ - --template-body file://deploy/service.yaml \ - --parameters file://deploy/parameters-dev.json --capabilities CAPABILITY_IAM -aws cloudformation wait stack-create-complete --stack-name forms-processor-v6-dev +CircleCI checks all branches. After quality and deployment tests pass, `develop` +releases to dev and `master` releases to production using `org-global` and the pinned +Topcoder AWS credential helper. The deployment registers a new task revision and +updates **forms-processor-v6** on **topcoder-infrastructure**. It does not create or +update CloudFormation stacks, services, networking, IAM roles, topics or databases. + +The existing service and ECR repository must already exist. `release.py` clones +registerable fields from the service's current task, preserving its roles, resource +limits, sidecars and other runtime configuration. It replaces the processor image, +managed SSM references and health check. Existing nonzero replica counts are kept; +a service at zero is started at one. A service with an in-progress rollout is rejected. +The script waits for healthy tasks of the requested revision and fails on rollback, +failed rollout or timeout. It records current/previous task ARNs in a release artifact. + +## Runtime parameter sources + +Values are injected by ECS at task startup; the release script reads only parameter +metadata. The task execution role must have `ssm:GetParameters` access to the selected +parameters and `kms:Decrypt` when a customer-managed key is used. + +| Settings | SSM source | +| --- | --- | +| `BUSAPI_URL` | `/config/forms-processor-v6/appvar/BUSAPI_URL` (required SecureString; prod `https://api.topcoder.com/v6`, dev `https://api.topcoder-dev.com/v6`) | +| `LETS_TALK_RECIPIENTS` | `/config/forms-processor-v6/appvar/LETS_TALK_RECIPIENTS` (SecureString; comma-separated addresses) | +| `LETS_TALK_SENDGRID_TEMPLATE_ID` | `/config/forms-processor-v6/appvar/LETS_TALK_SENDGRID_TEMPLATE_ID` (SecureString; active `d-` dynamic-template ID) | +| `DATABASE_URL`, `AUTH0_CLIENT_ID`, `AUTH0_CLIENT_SECRET` | `/config/forms-api-v6/appvar/` | +| `AUTH0_URL`, `AUTH0_AUDIENCE`, optional `AUTH0_PROXY_SERVER_URL` | Forms API appvars when present, otherwise `/config/common/global-appvar/` | +| `KAFKA_URL`, optional `KAFKA_SSL_ENABLED` | `/config/common/global-appvar/` | + +Only these named references are imported. The processor does not receive the API's +migration URL or unrelated shared secrets. If global Kafka TLS is absent, the current +task setting is preserved (runtime default false); ensure it matches the global broker +listeners. There are no processor SendGrid credentials: email-service-v6 owns delivery. + +The two email parameters may be absent during initial deployment. Their absence +produces empty runtime email settings; enabling the flow then defers email delivery +with `invalid_email_settings`. It never falls back to stale database recipients. +Malformed nonempty values fail startup without exposing values. Create both parameters +before the email release. **Redeploy after every parameter change** so new tasks read +new values and newly created parameters are included in the task definition. +The CI role needs ECR push/read, ECS describe/list/register/update, SSM metadata reads, +and permission to pass the existing task roles. No CloudFormation permissions are +needed. Accounts are guarded explicitly: dev `811668436784`, production `409275337247`. + +## Release and enable Let's Talk + +1. Deploy forms-api-v6 migrations, including `20260928020000_processor_flows` and + `20261013010000_lets_talk_processor_key`. The second fixes the seed's `lets-talk` + mismatch to the website's `lets_talk` key and reroutes retained unmatched events, + preserving existing delivery receipts and configuration. +2. Verify the global brokers are reachable from the existing processor security group, + the forms database permits the service, and `form.submitted` exists. The worker + disables Kafka topic auto-creation. Its group defaults to `forms-processor-v6`. +3. Set the processor SecureStrings above. The reused API M2M identity needs + `write:bus_api`. Configure an active SendGrid template using + [lets-talk-email.json](../examples/lets-talk-email.json) and the README variables. +4. Merge reviewed changes to develop, then master. The repository must be connected + to CircleCI with access to `org-global`. CircleCI builds the tested runtime image + and invokes `deploy/release.py production production-COMMIT-BUILD`. +5. Once healthy, inspect the pending-delivery query in the README. Enabling releases + matching queued submissions as well as future ones. Run the reviewed update in + [configure-lets-talk.sql](../examples/configure-lets-talk.sql) to enable the flow. +6. Publish the production form using the API's `scripts/publish-lets-talk.ts`, with + `FORMS_API_BASE_URL=https://api.topcoder.com/v6` and a production admin token. Release + the website migration only after the public schema and website CORS origin pass. +7. Submit one clearly identified test through `/lets-talk`. Follow its `submissionId` + from the API receipt into `form.submitted`, `ProcessorEvent`, `ProcessorDelivery`, + and the `external.action.email` envelope. Verify email-service-v6 accepts the + SendGrid request and the designated recipients receive correctly filled emails. + `deliveredAt` alone proves only Bus API acceptance, not inbox delivery. + +For a manual release of a reviewed image with temporary credentials: + +```sh docker build -t forms-processor-v6:candidate . -python3 deploy/release.py dev dev-UNIQUE-RELEASE-ID +python3 deploy/release.py production production-UNIQUE-RELEASE-ID ``` -The release script requires Python 3 with boto3, Docker, and an existing stable -stack. Use inherited temporary AWS credentials; never write them into images or -the repo. -Production release requires `PRODUCTION_AWS_ACCOUNT_ID` in the CircleCI project or -context as an explicit account guard. Dev is restricted to account 811668436784. -CloudFormation parameters intentionally have no dev-network defaults. Each AWS -account gets one fixed service/repository name; do not bootstrap two environments -in the same account under these names. - -CircleCI uses `org-global`, the pinned `tc-deploy-scripts` v1.4.20 credential helper, -and serial deployments per environment. The GitHub repository must be connected -to CircleCI and permitted to use the context. Deployment does not auto-enable any -email flow. Update the database settings and enable explicitly when ready. - -To roll back application code, update ImageTag to a previously verified ECR tag in -CloudFormation, preserving the other parameters. Keep additive DB migrations. -Do not deploy through the shared master_deploy.sh after adopting this stack, because -that would change the ECS service outside CloudFormation. Changes to new template -parameters without defaults require explicit bootstrap values. Removed parameters -are filtered when updating the template; new parameters with defaults use those defaults. - -API migration source is in the sibling forms-api-v6 repository, not owned or -silently rerun by this worker. The two repository changes must be released in order. +Rollback by updating the existing service to the prior task ARN recorded in the +release artifact. Keep additive API migrations. Disable the flow first if recipient +or template settings need correction; a send already in progress may finish. + +## Historical CloudFormation bootstrap + +`service.yaml` and `parameters-dev.json` describe the original dev bootstrap, retained +for reference. They are no longer applied by CI/CD. Subsequent infrastructure changes +must preserve the live task reference and desired count: applying an old stack image +or the old `release.py` procedure can revert a newer ECS release. The dev execution +role from that bootstrap needs access to the new processor SSM paths before its first +new release. Provision infrastructure separately if the production service is missing. diff --git a/deploy/release.py b/deploy/release.py index 8be630d..4a8ade9 100644 --- a/deploy/release.py +++ b/deploy/release.py @@ -1,10 +1,11 @@ #!/usr/bin/env python3 -"""Push a prebuilt image and update the existing CloudFormation stack. +"""Release a prebuilt image to an existing ECS service, without provisioning infrastructure. -Usage: release.py dev|production IMAGE_TAG [LOCAL_IMAGE]. Uses inherited AWS -credentials and Docker. Run the forms-api-v6 migration before the first release. +Usage: release.py dev|production IMAGE_TAG [LOCAL_IMAGE]. Uses inherited temporary +AWS credentials, boto3 and Docker. API migrations must precede this release. """ import base64 +from copy import deepcopy import json import os from pathlib import Path @@ -12,63 +13,173 @@ import subprocess import sys import time + import boto3 +APP = 'forms-processor-v6' +CLUSTER = 'topcoder-infrastructure' +ACCOUNTS = {'dev': '811668436784', 'production': '409275337247'} + + +def read_service(ecs): + """Read the fixed ECS service using ecs; return active service or raise on missing/invalid state.""" + result = ecs.describe_services(cluster=CLUSTER, services=[APP]) + if result.get('failures') or len(result.get('services', [])) != 1: + raise RuntimeError('Existing forms-processor-v6 ECS service is required') + service = result['services'][0] + if service['status'] != 'ACTIVE': + raise RuntimeError('Existing ECS service must be active') + return service + + +def secret_references(ssm): + """Resolve allowlisted SSM names for ECS injection, without decrypting any values. + + Takes an SSM client; returns name/valueFrom pairs. API Auth0 settings take + precedence over shared defaults. Kafka uses globals. Missing optional sales + settings permit a disabled-flow deployment. Raises for absent required values + or sales/Bus parameters that are not SecureStrings; AWS errors propagate. + """ + prefixes = {'api': '/config/forms-api-v6/appvar', + 'app': f'/config/{APP}/appvar', + 'common': '/config/common/global-appvar'} + available = {} + for prefix in prefixes.values(): + for page in ssm.get_paginator('describe_parameters').paginate( + ParameterFilters=[{'Key': 'Name', 'Option': 'BeginsWith', 'Values': [prefix + '/']}]): + available.update({p['Name']: p['Type'] for p in page['Parameters']}) + sources = {'DATABASE_URL': ['api'], 'BUSAPI_URL': ['app'], + 'AUTH0_CLIENT_ID': ['api'], 'AUTH0_CLIENT_SECRET': ['api'], + 'AUTH0_URL': ['api', 'common'], 'AUTH0_AUDIENCE': ['api', 'common'], + 'AUTH0_PROXY_SERVER_URL': ['api', 'common'], + 'KAFKA_URL': ['common'], 'KAFKA_SSL_ENABLED': ['common'], + 'LETS_TALK_RECIPIENTS': ['app'], 'LETS_TALK_SENDGRID_TEMPLATE_ID': ['app']} + optional = {'AUTH0_PROXY_SERVER_URL', 'KAFKA_SSL_ENABLED', + 'LETS_TALK_RECIPIENTS', 'LETS_TALK_SENDGRID_TEMPLATE_ID'} + references = [] + for name, candidates in sources.items(): + path = next((prefixes[source] + '/' + name for source in candidates + if prefixes[source] + '/' + name in available), None) + if path is None: + if name not in optional: + raise RuntimeError('Missing required SSM parameter: ' + name) + continue + if name in ('BUSAPI_URL', 'LETS_TALK_RECIPIENTS', 'LETS_TALK_SENDGRID_TEMPLATE_ID'): + if available[path] != 'SecureString': + raise RuntimeError('SecureString required: ' + path) + references.append({'name': name, 'valueFrom': path}) + return references + + +def task_request(ecs, current, image, references): + """Clone the live task's registerable fields and replace the processor image/managed secrets. + + Inputs are the ECS client, described task, ECR image URI and SSM references. + Returns a register-task-definition request, preserving roles/resources/sidecars. + Raises if family or application container identity differs from the fixed service. + """ + if current['family'] != APP: + raise RuntimeError('Unexpected task family') + allowed = ecs.meta.service_model.operation_model('RegisterTaskDefinition').input_shape.members + request = {key: deepcopy(value) for key, value in current.items() if key in allowed} + containers = [c for c in request['containerDefinitions'] if c['name'] == APP] + if len(containers) != 1: + raise RuntimeError('Expected one forms-processor-v6 container') + container = containers[0] + managed = {'DATABASE_URL', 'BUSAPI_URL', 'AUTH0_CLIENT_ID', 'AUTH0_CLIENT_SECRET', + 'AUTH0_URL', 'AUTH0_AUDIENCE', 'AUTH0_PROXY_SERVER_URL', 'KAFKA_URL', + 'LETS_TALK_RECIPIENTS', 'LETS_TALK_SENDGRID_TEMPLATE_ID'} + # Preserve an existing TLS setting if the global store does not define it. + managed.update(ref['name'] for ref in references) + container['environment'] = [e for e in container.get('environment', []) if e['name'] not in managed] + container['secrets'] = [s for s in container.get('secrets', []) if s['name'] not in managed] + references + container['image'] = image + container['healthCheck'] = { + 'command': ['CMD', 'node', '-e', + "fetch('http://127.0.0.1:'+(process.env.PORT||3000)+'/health').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))"], + 'interval': 30, 'timeout': 5, 'retries': 3, 'startPeriod': 60, + } + return request + + +def verify_service(ecs, definition_arn): + """Observe one rollout using ecs and the requested ARN; return service when tasks are healthy. + + Returns None while converging; raises on rollback/replacement/failed rollout or + AWS errors. Used by the bounded release waiter, including activation from zero. + """ + service = read_service(ecs) + if service['taskDefinition'] != definition_arn: + raise RuntimeError('ECS rolled back or replaced the requested task definition') + deployments = service['deployments'] + if any(d.get('rolloutState') == 'FAILED' for d in deployments): + raise RuntimeError('ECS deployment failed') + if (len(deployments) != 1 or deployments[0].get('rolloutState') != 'COMPLETED' + or service['runningCount'] != service['desiredCount'] or service['desiredCount'] < 1 + or service['pendingCount']): + return None + arns = [] + for page in ecs.get_paginator('list_tasks').paginate( + cluster=CLUSTER, serviceName=APP, desiredStatus='RUNNING'): + arns.extend(page['taskArns']) + tasks = [] + for start in range(0, len(arns), 100): + tasks.extend(ecs.describe_tasks(cluster=CLUSTER, tasks=arns[start:start + 100])['tasks']) + healthy = [t for t in tasks if t['taskDefinitionArn'] == definition_arn + and t.get('healthStatus') == 'HEALTHY' and t['lastStatus'] == 'RUNNING'] + return service if len(healthy) >= service['desiredCount'] else None + def main(): - """Validate arguments/account, push an immutable tag, promote the stack, and record verified ECS state. + """Read CLI arguments, verify account/service, push image, register a revision and update ECS. - Returns None; raises on AWS/Docker failure, unstable stacks, or failed promotion. - Passwords are passed on stdin and never logged. Existing network parameters are preserved. + Returns None after verified deployment and a non-secret release artifact. Raises + on invalid arguments/account, missing existing infrastructure, AWS/Docker errors, + concurrent service change, rollback or a 30-minute timeout. Never updates stacks, + creates services, modifies networking/IAM, runs migrations, or enables email flows. """ - if len(sys.argv) not in (3, 4) or sys.argv[1] not in ('dev', 'production'): + if len(sys.argv) not in (3, 4) or sys.argv[1] not in ACCOUNTS: raise RuntimeError(__doc__) environment, tag = sys.argv[1:3] if not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,110}', tag): raise RuntimeError('Invalid image tag') - local_image = sys.argv[3] if len(sys.argv) == 4 else 'forms-processor-v6:candidate' session = boto3.Session(region_name=os.environ.get('AWS_REGION', 'us-east-1')) - expected = '811668436784' if environment == 'dev' else os.environ.get('PRODUCTION_AWS_ACCOUNT_ID') - if not expected or session.client('sts').get_caller_identity()['Account'] != expected: - raise RuntimeError('AWS account mismatch; production requires PRODUCTION_AWS_ACCOUNT_ID') - cfn, ecr, ecs = [session.client(name) for name in ('cloudformation', 'ecr', 'ecs')] - stack_name = f'forms-processor-v6-{environment}' - stack = cfn.describe_stacks(StackName=stack_name)['Stacks'][0] - if stack['StackStatus'] not in ('CREATE_COMPLETE', 'UPDATE_COMPLETE', 'UPDATE_ROLLBACK_COMPLETE'): - raise RuntimeError('Stack is not stable; inspect before retrying') - settings = {p['ParameterKey']: p['ParameterValue'] for p in stack['Parameters']} - if settings['Environment'] != environment: - raise RuntimeError('Stack environment mismatch') - outputs = {o['OutputKey']: o['OutputValue'] for o in stack['Outputs']} - repository = outputs['RepositoryUri'] + account = session.client('sts').get_caller_identity()['Account'] + if account != ACCOUNTS[environment]: + raise RuntimeError('AWS account does not match requested environment') + ecs, ecr, ssm = [session.client(name) for name in ('ecs', 'ecr', 'ssm')] + service = read_service(ecs) + if len(service['deployments']) != 1 or service['deployments'][0].get('rolloutState') != 'COMPLETED': + raise RuntimeError('Existing ECS deployment must be stable before releasing') + current = ecs.describe_task_definition(taskDefinition=service['taskDefinition'])['taskDefinition'] + references = secret_references(ssm) + repository = ecr.describe_repositories(repositoryNames=[APP])['repositories'][0]['repositoryUri'] + image = f'{repository}:{tag}' + request = task_request(ecs, current, image, references) authorization = ecr.get_authorization_token()['authorizationData'][0] user, password = base64.b64decode(authorization['authorizationToken']).decode().split(':', 1) - subprocess.run(['docker', 'login', '--username', user, '--password-stdin', authorization['proxyEndpoint']], input=password, text=True, check=True, stdout=subprocess.DEVNULL) - image = f'{repository}:{tag}' + subprocess.run(['docker', 'login', '--username', user, '--password-stdin', authorization['proxyEndpoint']], + input=password, text=True, check=True, stdout=subprocess.DEVNULL) + local_image = sys.argv[3] if len(sys.argv) == 4 else APP + ':candidate' subprocess.run(['docker', 'tag', local_image, image], check=True) subprocess.run(['docker', 'push', image], check=True) - template = Path('deploy/service.yaml').read_text() - # Preserve only parameters still declared by the new template; additions use defaults. - declared = {p['ParameterKey'] for p in cfn.validate_template(TemplateBody=template)['Parameters']} - parameters = [{'ParameterKey': key, 'UsePreviousValue': True} for key in settings if key in declared and key not in ('ImageTag', 'DesiredCount')] - parameters += [{'ParameterKey': 'ImageTag', 'ParameterValue': tag}, {'ParameterKey': 'DesiredCount', 'ParameterValue': str(max(1, int(settings['DesiredCount'])))}] - cfn.update_stack(StackName=stack_name, TemplateBody=template, Parameters=parameters, Capabilities=['CAPABILITY_IAM']) + if read_service(ecs)['taskDefinition'] != service['taskDefinition']: + raise RuntimeError('Service changed during image push; retry after inspecting concurrent release') + definition_arn = ecs.register_task_definition(**request)['taskDefinition']['taskDefinitionArn'] + ecs.update_service(cluster=CLUSTER, service=APP, taskDefinition=definition_arn, + desiredCount=max(1, service['desiredCount'])) deadline = time.monotonic() + 1800 while time.monotonic() < deadline: - state = cfn.describe_stacks(StackName=stack_name)['Stacks'][0]['StackStatus'] - if state == 'UPDATE_COMPLETE': + retained = verify_service(ecs, definition_arn) + if retained: break - if state not in ('UPDATE_IN_PROGRESS', 'UPDATE_COMPLETE_CLEANUP_IN_PROGRESS'): - raise RuntimeError('Deployment failed or rolled back: ' + state) - print('Waiting for CloudFormation: ' + state, flush=True) + print('Waiting for healthy requested ECS revision', flush=True) time.sleep(15) else: - raise RuntimeError('Deployment timed out; inspect existing operation before retrying') - service = ecs.describe_services(cluster=settings['ClusterName'], services=['forms-processor-v6'])['services'][0] - task = ecs.describe_task_definition(taskDefinition=service['taskDefinition'])['taskDefinition'] - if task['containerDefinitions'][0]['image'] != image or service['runningCount'] != service['desiredCount']: - raise RuntimeError('ECS did not retain the requested release') - evidence = {'stack': stack_name, 'image': image, 'taskDefinition': service['taskDefinition'], 'runningCount': service['runningCount']} + raise RuntimeError('ECS deployment timed out; inspect before retrying') + evidence = {'environment': environment, 'cluster': CLUSTER, 'service': APP, 'image': image, + 'taskDefinition': definition_arn, 'previousTaskDefinition': service['taskDefinition'], + 'runningCount': retained['runningCount']} Path(f'deploy/release-{environment}.json').write_text(json.dumps(evidence, indent=2) + '\n') print(json.dumps(evidence, indent=2)) diff --git a/deploy/tests/test_release.py b/deploy/tests/test_release.py new file mode 100644 index 0000000..da11fd9 --- /dev/null +++ b/deploy/tests/test_release.py @@ -0,0 +1,108 @@ +"""Exercise rollout guards and secret-reference selection without contacting AWS.""" +import importlib.util +from pathlib import Path +from types import SimpleNamespace +import unittest +from unittest.mock import Mock, patch + +import boto3 + +spec = importlib.util.spec_from_file_location('release', Path(__file__).parents[1] / 'release.py') +release = importlib.util.module_from_spec(spec) +spec.loader.exec_module(release) + + +class ReleaseTests(unittest.TestCase): + """Verify that releases update existing ECS tasks and cannot cross environment boundaries.""" + + def test_missing_service_is_rejected(self): + """A missing service raises before release can provision anything; no inputs or return value.""" + ecs = Mock() + ecs.describe_services.return_value = {'services': [], 'failures': [{'reason': 'MISSING'}]} + with self.assertRaisesRegex(RuntimeError, 'Existing'): + release.read_service(ecs) + ecs.update_service.assert_not_called() + + def test_secret_sources_and_types(self): + """Select API identity, global Kafka and processor Bus/email SecureStrings; rejects invalid metadata.""" + entries = { + '/config/forms-api-v6/appvar': ['DATABASE_URL', 'AUTH0_CLIENT_ID', 'AUTH0_CLIENT_SECRET', 'AUTH0_URL'], + '/config/forms-processor-v6/appvar': ['BUSAPI_URL', 'LETS_TALK_RECIPIENTS', 'LETS_TALK_SENDGRID_TEMPLATE_ID'], + '/config/common/global-appvar': ['AUTH0_URL', 'AUTH0_AUDIENCE', 'AUTH0_PROXY_SERVER_URL', 'KAFKA_URL', 'KAFKA_SSL_ENABLED'], + } + ssm = Mock() + pages = [[{'Parameters': [{'Name': path + '/' + name, 'Type': 'SecureString'} for name in names]}] + for path, names in entries.items()] + ssm.get_paginator.return_value.paginate.side_effect = pages + refs = {r['name']: r['valueFrom'] for r in release.secret_references(ssm)} + self.assertEqual(refs['AUTH0_URL'], '/config/forms-api-v6/appvar/AUTH0_URL') + self.assertEqual(refs['KAFKA_URL'], '/config/common/global-appvar/KAFKA_URL') + self.assertEqual(refs['BUSAPI_URL'], '/config/forms-processor-v6/appvar/BUSAPI_URL') + self.assertEqual(len(refs), 11) + ssm.get_parameter.assert_not_called() + pages[1][0]['Parameters'][0]['Type'] = 'String' + ssm.get_paginator.return_value.paginate.side_effect = pages + with self.assertRaisesRegex(RuntimeError, 'SecureString'): + release.secret_references(ssm) + ssm.get_paginator.return_value.paginate.side_effect = [[{'Parameters': []}]] * 3 + with self.assertRaisesRegex(RuntimeError, 'Missing required'): + release.secret_references(ssm) + + def test_clone_preserves_roles_resources_and_removes_stale_literals(self): + """Cloning returns a registerable copy with exact secret overrides and unchanged sidecars/roles.""" + ecs = boto3.client('ecs', region_name='us-east-1', aws_access_key_id='test', aws_secret_access_key='test') + current = {'family': release.APP, 'taskDefinitionArn': 'old', 'revision': 7, + 'executionRoleArn': 'execution', 'taskRoleArn': 'task', 'cpu': '256', 'memory': '512', + 'networkMode': 'awsvpc', 'requiresCompatibilities': ['FARGATE'], + 'containerDefinitions': [{'name': release.APP, 'image': 'old', 'readonlyRootFilesystem': True, + 'environment': [{'name': 'BUSAPI_URL', 'value': 'old'}, {'name': 'KAFKA_SSL_ENABLED', 'value': 'true'}], + 'secrets': [{'name': 'LETS_TALK_RECIPIENTS', 'valueFrom': 'old'}, {'name': 'OTHER', 'valueFrom': '/other'}]}, + {'name': 'sidecar', 'image': 'sidecar'}]} + refs = [{'name': 'BUSAPI_URL', 'valueFrom': '/config/forms-processor-v6/appvar/BUSAPI_URL'}] + result = release.task_request(ecs, current, 'new', refs) + self.assertNotIn('revision', result) + self.assertNotIn('taskDefinitionArn', result) + for key in ['executionRoleArn', 'taskRoleArn', 'cpu', 'memory', 'networkMode', 'requiresCompatibilities']: + self.assertEqual(result[key], current[key]) + container = result['containerDefinitions'][0] + self.assertEqual(container['environment'], [{'name': 'KAFKA_SSL_ENABLED', 'value': 'true'}]) + self.assertEqual(container['secrets'], [{'name': 'OTHER', 'valueFrom': '/other'}] + refs) + self.assertEqual(container['image'], 'new') + self.assertEqual(current['containerDefinitions'][0]['image'], 'old') + self.assertEqual(result['containerDefinitions'][1], current['containerDefinitions'][1]) + self.assertIn('healthCheck', container) + + def test_rollback_and_unhealthy_tasks_are_not_success(self): + """Only healthy tasks of the requested revision finish verification; rollback raises.""" + ecs = Mock() + service = {'status': 'ACTIVE', 'taskDefinition': 'new', 'desiredCount': 1, 'runningCount': 1, + 'pendingCount': 0, 'deployments': [{'rolloutState': 'COMPLETED'}]} + ecs.describe_services.return_value = {'services': [service]} + ecs.get_paginator.return_value.paginate.return_value = [{'taskArns': ['task']}] + task = {'taskDefinitionArn': 'new', 'healthStatus': 'UNHEALTHY', 'lastStatus': 'RUNNING'} + ecs.describe_tasks.return_value = {'tasks': [task]} + self.assertIsNone(release.verify_service(ecs, 'new')) + task['healthStatus'] = 'HEALTHY' + self.assertEqual(release.verify_service(ecs, 'new'), service) + service['deployments'][0]['rolloutState'] = 'FAILED' + with self.assertRaisesRegex(RuntimeError, 'failed'): + release.verify_service(ecs, 'new') + service['taskDefinition'] = 'old' + with self.assertRaisesRegex(RuntimeError, 'rolled back'): + release.verify_service(ecs, 'new') + + def test_wrong_account_stops_before_mutation(self): + """Production with dev credentials raises before ECR, ECS, SSM or Docker are called.""" + session = Mock(region_name='us-east-1') + session.client.return_value.get_caller_identity.return_value = {'Account': release.ACCOUNTS['dev']} + with patch.object(release.boto3, 'Session', return_value=session), \ + patch.object(release.sys, 'argv', ['release.py', 'production', 'candidate']), \ + patch.object(release.subprocess, 'run') as run: + with self.assertRaisesRegex(RuntimeError, 'account'): + release.main() + session.client.assert_called_once_with('sts') + run.assert_not_called() + + +if __name__ == '__main__': + unittest.main() diff --git a/examples/configure-lets-talk.sql b/examples/configure-lets-talk.sql index ce1ee21..da337a6 100644 --- a/examples/configure-lets-talk.sql +++ b/examples/configure-lets-talk.sql @@ -1,15 +1,15 @@ --- Replace TBD recipients/template, verify Bus API M2M configuration, then execute as an authorized DB admin. --- fromEmail is optional: use null for the email-service-v6 default, or a verified sender override. --- Enabled remains false here deliberately. Enable with a separate UPDATE after configuration is complete. +-- Run only after the API form-key migration, processor deployment, and SSM setup. +-- Recipients/template come exclusively from these SecureStrings at task startup: +-- /config/forms-processor-v6/appvar/LETS_TALK_RECIPIENTS +-- /config/forms-processor-v6/appvar/LETS_TALK_SENDGRID_TEMPLATE_ID +-- Redeploy after setting them. The sender comes from email-service-v6's default. +-- Review queued deliveries before enabling: they will also be sent. UPDATE forms."ProcessorFlow" -SET settings = jsonb_build_object( - 'recipients', jsonb_build_array('TBD'), - 'templateId', 'TBD', - 'fromEmail', NULL, - 'fromName', 'Topcoder' - ), - "updatedAt" = now() -WHERE id = 'lets-talk-sales-email'; +SET enabled = true, "updatedAt" = now() +WHERE id = 'lets-talk-sales-email' AND "formKey" = 'lets_talk' + AND action = 'sendgrid-email' +RETURNING id, "formKey", enabled; --- UPDATE forms."ProcessorFlow" SET enabled = true, "updatedAt" = now() +-- Pause pending sends when necessary (in-progress sends may complete): +-- UPDATE forms."ProcessorFlow" SET enabled = false, "updatedAt" = now() -- WHERE id = 'lets-talk-sales-email'; diff --git a/examples/lets-talk-email.json b/examples/lets-talk-email.json new file mode 100644 index 0000000..4fd819d --- /dev/null +++ b/examples/lets-talk-email.json @@ -0,0 +1,97 @@ +{ + "topic": "external.action.email", + "originator": "forms-processor-v6", + "timestamp": "2026-10-13T00:00:00.000Z", + "mime-type": "application/json", + "key": "25fa4e9d-b6bb-4013-983b-20d243a22ca2", + "payload": { + "recipients": [ + "sales@example.com", + "associate@example.com" + ], + "sendgrid_template_id": "d-00000000000000000000000000000000", + "version": "v3", + "data": { + "submissionId": "25fa4e9d-b6bb-4013-983b-20d243a22ca2", + "formKey": "lets_talk", + "version": 1, + "submittedAt": "2026-10-13T00:00:00.000Z", + "memberId": null, + "sourcePage": "/lets-talk", + "answers": { + "interested_in": "crowdsourcing", + "firstname": "Alex", + "lastname": "Example", + "email": "alex@example.com", + "company": "Example Company", + "briefly_describe_your_inquiry": "TEST submission — please confirm the production form email.", + "leadsource": "landing_page", + "source__c": "/lets-talk", + "lead_source_for_campaign__c": "NewTCLet'sTalk", + "primary_campaign__c": "NewTCLet'sTalk", + "insta_page_name__c": "New Topcoder.com - Let's chat form", + "utm_medium__c": "", + "utm_campaign__c": "", + "utm_source__c": "" + }, + "fields": [ + { + "key": "interested_in", + "value": "crowdsourcing" + }, + { + "key": "firstname", + "value": "Alex" + }, + { + "key": "lastname", + "value": "Example" + }, + { + "key": "email", + "value": "alex@example.com" + }, + { + "key": "company", + "value": "Example Company" + }, + { + "key": "briefly_describe_your_inquiry", + "value": "TEST submission — please confirm the production form email." + }, + { + "key": "leadsource", + "value": "landing_page" + }, + { + "key": "source__c", + "value": "/lets-talk" + }, + { + "key": "lead_source_for_campaign__c", + "value": "NewTCLet'sTalk" + }, + { + "key": "primary_campaign__c", + "value": "NewTCLet'sTalk" + }, + { + "key": "insta_page_name__c", + "value": "New Topcoder.com - Let's chat form" + }, + { + "key": "utm_medium__c", + "value": "" + }, + { + "key": "utm_campaign__c", + "value": "" + }, + { + "key": "utm_source__c", + "value": "" + } + ] + } + } +} diff --git a/src/config.ts b/src/config.ts index 8a104fc..9253bc1 100644 --- a/src/config.ts +++ b/src/config.ts @@ -19,10 +19,27 @@ const schema = z.object({ AUTH0_CLIENT_ID: z.string().optional(), AUTH0_CLIENT_SECRET: z.string().optional(), AUTH0_PROXY_SERVER_URL: z.string().optional(), + LETS_TALK_RECIPIENTS: z + .string() + .trim() + .default("") + .transform((value) => + value ? value.split(",").map((email) => email.trim().toLowerCase()) : [], + ) + .pipe(z.array(z.email()).max(1000)) + .transform((emails) => [...new Set(emails)]), + LETS_TALK_SENDGRID_TEMPLATE_ID: z + .string() + .trim() + .regex(/^$|^d-[a-f0-9]{32}$/) + .default(""), PORT: z.coerce.number().int().min(1).max(65535).default(3000), }); export type Config = z.infer & { busApi?: BusConfig }; -/** Validate environment settings at startup; returns config or throws with names only, never values. */ +/** Validate startup environment (defaults to process.env), including comma-separated sales recipients. + * Returns normalized worker/Bus settings; missing email settings leave delivery deferred. + * Throws with invalid variable names only, never secret values; used by main before opening connections. + */ export function config(env: NodeJS.ProcessEnv = process.env): Config { const parsed = schema.safeParse(env); if (!parsed.success) diff --git a/src/main.ts b/src/main.ts index d11efe1..bfd97f1 100644 --- a/src/main.ts +++ b/src/main.ts @@ -7,7 +7,9 @@ import { config } from "./config.js"; import { consume } from "./consumer.js"; import { Store, log } from "./store.js"; -/** Start the database inbox worker, Kafka consumer, and private health endpoint; propagate startup failures. */ +/** Start the database inbox worker, Kafka consumer, and private health endpoint with environment-owned sales settings. + * Takes no arguments and resolves after shutdown; startup/configuration/database failures propagate to the fatal handler. + */ async function main(): Promise { const settings = config(); const pool = new Pool({ @@ -28,6 +30,15 @@ async function main(): Promise { ), ], ]), + new Map([ + [ + "lets-talk-sales-email", + { + recipients: settings.LETS_TALK_RECIPIENTS, + templateId: settings.LETS_TALK_SENDGRID_TEMPLATE_ID, + }, + ], + ]), ); await store.ready(); const shutdown = new AbortController(); diff --git a/src/store.ts b/src/store.ts index 1d6ad65..a978b5f 100644 --- a/src/store.ts +++ b/src/store.ts @@ -17,12 +17,20 @@ export function retrySeconds(attempts: number): number { return Math.min(3600, 5 * 2 ** Math.min(attempts, 10)); } -/** Owns the durable inbox, routing, and per-flow delivery receipts in the API-managed forms schema. */ +/** Owns the durable inbox, routing, and per-flow receipts in the API-managed forms schema. + * main supplies actions and optional environment-owned settings for specific flows. + */ export class Store { - /** Accepts an existing PostgreSQL pool and action registry. Does not open connections or throw. */ + /** Accepts a PostgreSQL pool, action registry, and optional settings replacements keyed by flow ID. + * Returns a store without opening connections or throwing. main uses replacements for SSM-owned sales settings. + */ constructor( readonly pool: Pool, private readonly actions: ReadonlyMap, + private readonly settingsOverrides: ReadonlyMap< + string, + unknown + > = new Map(), ) {} /** Persist a validated submission before Kafka acknowledgement; duplicate IDs are no-ops. Throws on DB failure. */ @@ -108,6 +116,7 @@ export class Store { } /** Deliver one pending enabled flow under a row lock; records success or schedules a retry. + * Takes no input; environment settings replace DB settings only for explicitly configured flow IDs. * Returns whether work was found. Action failures are sanitized; DB failures propagate. * A crash after Bus API acceptance but before commit can resend (at-least-once delivery). */ @@ -125,7 +134,12 @@ export class Store { try { const action = this.actions.get(row.action); if (!action) throw new ActionError("unknown_action"); - await action.execute(submissionSchema.parse(row.payload), row.settings); + await action.execute( + submissionSchema.parse(row.payload), + this.settingsOverrides.has(row.flowId) + ? this.settingsOverrides.get(row.flowId) + : row.settings, + ); } catch (error) { const code = error instanceof ActionError ? error.message : "action_failed"; diff --git a/test/config.test.ts b/test/config.test.ts new file mode 100644 index 0000000..121e76a --- /dev/null +++ b/test/config.test.ts @@ -0,0 +1,37 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { config } from "../src/config.js"; +const base = { + DATABASE_URL: "postgresql://localhost/test", + KAFKA_URL: "localhost:9092", +}; + +test("sales settings normalize comma-separated recipients and leave unconfigured delivery empty", () => { + assert.deepEqual(config(base).LETS_TALK_RECIPIENTS, []); + assert.equal(config(base).LETS_TALK_SENDGRID_TEMPLATE_ID, ""); + const settings = config({ + ...base, + LETS_TALK_RECIPIENTS: + " sales@example.com,OTHER@example.com,SALES@example.com ", + }); + assert.deepEqual(settings.LETS_TALK_RECIPIENTS, [ + "sales@example.com", + "other@example.com", + ]); +}); +test("invalid sales settings fail startup without disclosing values", () => { + for (const [key, value] of [ + ["LETS_TALK_RECIPIENTS", "private@example.com,"], + ["LETS_TALK_RECIPIENTS", "private@example.com,invalid"], + ["LETS_TALK_SENDGRID_TEMPLATE_ID", "private-template"], + ]) { + assert.throws( + () => config({ ...base, [key!]: value }), + (error: unknown) => { + assert.ok(error instanceof Error); + assert.match(error.message, new RegExp(`^Invalid configuration: ${key}(?:\\.\\d+)?$`)); + return true; + }, + ); + } +}); diff --git a/test/fixtures.ts b/test/fixtures.ts index 1eb12d2..01e321e 100644 --- a/test/fixtures.ts +++ b/test/fixtures.ts @@ -1,7 +1,7 @@ import type { Submission } from "../src/contracts.js"; export const event: Submission = { submissionId: "25fa4e9d-b6bb-4013-983b-20d243a22ca2", - formKey: "lets-talk", + formKey: "lets_talk", version: 1, submittedAt: "2026-09-28T00:00:00.000Z", memberId: null, diff --git a/test/fixtures/lets-talk-key.sql b/test/fixtures/lets-talk-key.sql new file mode 100644 index 0000000..e06a071 --- /dev/null +++ b/test/fixtures/lets-talk-key.sql @@ -0,0 +1,15 @@ +-- The published form key is lets_talk; preserve the flow's settings and enabled state. +UPDATE forms."ProcessorFlow" +SET "formKey" = 'lets_talk', "updatedAt" = CURRENT_TIMESTAMP +WHERE id = 'lets-talk-sales-email' AND "formKey" = 'lets-talk'; + +-- Retry routing retained submissions that previously missed this flow. Existing +-- delivery receipts stay intact, so already accepted emails are not replayed. +UPDATE forms."ProcessorEvent" e +SET "routedAt" = NULL, "nextAttemptAt" = CURRENT_TIMESTAMP +WHERE e."formKey" = 'lets_talk' + AND EXISTS (SELECT 1 FROM forms."ProcessorFlow" f + WHERE f.id = 'lets-talk-sales-email' AND f."formKey" = 'lets_talk') + AND NOT EXISTS (SELECT 1 FROM forms."ProcessorDelivery" d + WHERE d."submissionId" = e."submissionId" + AND d."flowId" = 'lets-talk-sales-email'); diff --git a/test/integration/store.test.ts b/test/integration/store.test.ts index 1c5d38f..b8360db 100644 --- a/test/integration/store.test.ts +++ b/test/integration/store.test.ts @@ -37,6 +37,12 @@ before(async () => { "utf8", ), ); + await pool.query( + await readFile( + new URL("../fixtures/lets-talk-key.sql", import.meta.url), + "utf8", + ), + ); }); after(async () => { await pool.end(); @@ -80,7 +86,7 @@ test("concurrent workers claim each delivery only once", async () => { test("failed delivery retries later and preserves successful per-flow receipts", async () => { await pool.query('UPDATE forms."ProcessorFlow" SET enabled = true'); await pool.query( - "INSERT INTO forms.\"ProcessorFlow\" (id, \"formKey\", enabled, action) VALUES ('second', 'lets-talk', true, 'sendgrid-email')", + "INSERT INTO forms.\"ProcessorFlow\" (id, \"formKey\", enabled, action) VALUES ('second', 'lets_talk', true, 'sendgrid-email')", ); await store.enqueue(event); await store.routeOne(); @@ -141,10 +147,81 @@ test("re-routing preserves successful receipts and creates newly added matching await store.routeOne(); await store.deliverOne(); await pool.query( - "INSERT INTO forms.\"ProcessorFlow\" (id, \"formKey\", enabled, action) VALUES ('second', 'lets-talk', true, 'sendgrid-email')", + "INSERT INTO forms.\"ProcessorFlow\" (id, \"formKey\", enabled, action) VALUES ('second', 'lets_talk', true, 'sendgrid-email')", ); await pool.query('UPDATE forms."ProcessorEvent" SET "routedAt" = NULL'); await store.routeOne(); await store.deliverOne(); assert.equal(calls, 2); }); + +test("environment settings replace only the configured flow and missing settings never fall back to DB", async () => { + const seen: unknown[] = []; + const configured = new Store( + pool, + new Map([ + [ + "sendgrid-email", + { + execute: async (_event, settings) => { + seen.push(settings); + }, + }, + ], + ]), + new Map([["lets-talk-sales-email", { recipients: [], templateId: "" }]]), + ); + await pool.query( + 'UPDATE forms."ProcessorFlow" SET enabled = true, settings = $1', + [JSON.stringify({ recipients: ["stale@example.com"] })], + ); + await pool.query( + "INSERT INTO forms.\"ProcessorFlow\" (id, \"formKey\", enabled, action, settings) VALUES ('second', 'lets_talk', true, 'sendgrid-email', '{\"custom\":true}')", + ); + await configured.enqueue(event); + await configured.routeOne(); + await configured.deliverOne(); + await configured.deliverOne(); + assert.deepEqual(seen, [ + { recipients: [], templateId: "" }, + { custom: true }, + ]); +}); + +test("form-key migration reroutes missed website events and preserves configuration and accepted receipts", async () => { + const second = { + ...event, + submissionId: "bf1d6716-26bf-4b66-aa23-9904f22d8b41", + }; + await pool.query('UPDATE forms."ProcessorFlow" SET enabled = true'); + await store.enqueue(event); + await store.routeOne(); + await store.deliverOne(); + await pool.query( + 'UPDATE forms."ProcessorFlow" SET "formKey" = \'lets-talk\'', + ); + await store.enqueue(second); + await store.routeOne(); + const before = ( + await pool.query( + 'SELECT enabled, settings, rules FROM forms."ProcessorFlow"', + ) + ).rows[0]; + await pool.query( + await readFile( + new URL("../fixtures/lets-talk-key.sql", import.meta.url), + "utf8", + ), + ); + const flow = ( + await pool.query( + 'SELECT "formKey", enabled, settings, rules FROM forms."ProcessorFlow"', + ) + ).rows[0]; + assert.deepEqual(flow, { ...before, formKey: "lets_talk" }); + assert.equal(await store.routeOne(), true); + assert.equal(await store.routeOne(), false); + assert.equal(await store.deliverOne(), true); + assert.equal(await store.deliverOne(), false); + assert.equal(calls, 2); +});