id: gdpr-dsar-orchestrator
namespace: company.team
description: |
Handle a GDPR data subject request end to end. Find a person's data in
Postgres, S3-compatible object storage and OpenSearch in parallel, then either
export it as a ZIP bundle (access request) or, after a named person approves,
erase it, verify nothing is left, and write a proof-of-erasure record.
labels:
team: privacy
use-case: gdpr-dsar
triggers:
- id: intake
type: io.kestra.plugin.core.trigger.Webhook
description: A privacy form or helpdesk POSTs {"request_id", "email", "type",
"dry_run"}. The body is mapped onto the typed inputs, so the email is
validated before anything runs.
key: "{{ secret('DSAR_WEBHOOK_KEY') }}"
inputs:
request_id: "{{ trigger.body.request_id }}"
subject_email: "{{ trigger.body.email }}"
request_type: "{{ trigger.body.type ?? 'ACCESS' }}"
dry_run: "{{ trigger.body.dry_run ?? true }}"
inputs:
- id: request_id
type: STRING
displayName: Request ID
description: Ticket or reference number of the request. Used for labels, the
audit record and duplicate detection.
- id: subject_email
type: EMAIL
displayName: Data subject email
- id: request_type
type: SELECT
displayName: Request type
values:
- ACCESS
- ERASURE
defaults: ACCESS
- id: dry_run
type: BOOL
displayName: Dry run
description: For erasure requests, only report what would be erased. Set to
false to erase after approval.
defaults: true
- id: bucket
type: STRING
displayName: Bucket
description: Bucket holding user files under customers/<customer id>/.
defaults: user-uploads
- id: s3_endpoint
type: STRING
displayName: S3 endpoint
description: S3 API endpoint. Use https://s3.<region>.amazonaws.com for AWS, or
your S3-compatible store such as SeaweedFS.
defaults: https://s3.us-east-1.amazonaws.com
- id: tickets_index
type: STRING
displayName: Support tickets index
defaults: support-tickets
tasks:
- id: label_request
type: io.kestra.plugin.core.execution.Labels
description: Tag the execution with the request, so every run for a request can
be found in the UI.
labels:
request_id: "{{ inputs.request_id }}"
request_type: "{{ inputs.request_type }}"
- id: skip_if_done
type: io.kestra.plugin.core.flow.If
description: Stop here if this request already completed, so webhook retries and
duplicate submissions do nothing.
condition: "{{ ((kv('dsar-' ~ (inputs.request_id | slugify),
errorOnMissing=false) ?? {}).status ?? '') == 'COMPLETED' }}"
then:
- id: already_completed
type: io.kestra.plugin.core.execution.Exit
description: End quietly; the request was handled by an earlier execution.
state: SUCCESS
- id: register
type: io.kestra.plugin.core.kv.Set
description: Open the audit record with the legal deadline, 30 days after receipt.
key: "dsar-{{ inputs.request_id | slugify }}"
kvType: JSON
value: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "{{ inputs.request_type }}", "status": "IN_PROGRESS", "received_at": "{{ execution.startDate }}", "due_by": "{{ execution.startDate | dateAdd(30, 'DAYS') | date('yyyy-MM-dd') }}", "execution_id": "{{ execution.id }}"}
- id: find_customer
type: io.kestra.plugin.jdbc.postgresql.Query
description: Look up the customer ID, which keys the user's files in object
storage. The email is bound as a parameter, never written into the SQL.
url: "{{ secret('PG_URL') }}"
username: "{{ secret('PG_USERNAME') }}"
password: "{{ secret('PG_PASSWORD') }}"
sql: SELECT id FROM customers WHERE lower(email) = lower(:email)
parameters:
email: "{{ inputs.subject_email }}"
fetchType: FETCH_ONE
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: remember_customer
type: io.kestra.plugin.core.flow.If
description: Keep the customer ID for this request, so a re-run after a partial
erasure still finds the files even though the email was already
anonymized.
condition: "{{ outputs.find_customer.size > 0 }}"
then:
- id: store_customer_id
type: io.kestra.plugin.core.kv.Set
description: Store the customer ID next to the audit record.
key: "dsar-{{ inputs.request_id | slugify }}-customer"
value: "{{ outputs.find_customer.row.id }}"
- id: subject
type: io.kestra.plugin.core.output.OutputValues
description: Resolve the customer ID from the lookup, or from the stored value
when the email was already erased by an earlier run.
values:
customer_id: "{{ outputs.find_customer.size > 0 ? outputs.find_customer.row.id :
(kv('dsar-' ~ (inputs.request_id | slugify) ~ '-customer',
errorOnMissing=false) ?? '') }}"
- id: discover
type: io.kestra.plugin.core.flow.Parallel
description: Search every system at once.
tasks:
- id: db_records
type: io.kestra.plugin.jdbc.postgresql.Query
description: Collect the customer row, orders and marketing consents as one JSON
document.
url: "{{ secret('PG_URL') }}"
username: "{{ secret('PG_USERNAME') }}"
password: "{{ secret('PG_PASSWORD') }}"
sql: |
SELECT CAST(json_build_object(
'customer', (SELECT row_to_json(c) FROM customers c WHERE lower(c.email) = lower(:email)),
'orders', COALESCE((SELECT json_agg(o ORDER BY o.created_at) FROM orders o JOIN customers c ON c.id = o.customer_id WHERE lower(c.email) = lower(:email)), CAST('[]' AS json)),
'marketing_consents', COALESCE((SELECT json_agg(m) FROM marketing_consents m JOIN customers c ON c.id = m.customer_id WHERE lower(c.email) = lower(:email)), CAST('[]' AS json))
) AS text) AS export,
(SELECT count(*) FROM orders o JOIN customers c ON c.id = o.customer_id WHERE lower(c.email) = lower(:email))
+ (SELECT count(*) FROM marketing_consents m JOIN customers c ON c.id = m.customer_id WHERE lower(c.email) = lower(:email))
+ (SELECT count(*) FROM customers c WHERE lower(c.email) = lower(:email)) AS row_count
parameters:
email: "{{ inputs.subject_email }}"
fetchType: FETCH_ONE
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: tickets
type: io.kestra.plugin.opensearch.Request
description: Find support tickets filed by or about the subject. Uses the
Request task because its body supports expressions.
connection:
hosts:
- "{{ secret('OPENSEARCH_URL') }}"
method: POST
endpoint: "/{{ inputs.tickets_index }}/_search"
body:
size: 1000
query:
term:
customer_email: "{{ inputs.subject_email | lower }}"
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: files
type: io.kestra.plugin.aws.s3.List
description: List the user's files. A subject without a customer record has no
files.
endpointOverride: "{{ inputs.s3_endpoint }}"
forcePathStyle: true
region: us-east-1
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ inputs.bucket }}"
prefix: "customers/{{ outputs.subject.values.customer_id != '' ?
outputs.subject.values.customer_id : 'no-customer-record' }}/"
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: findings
type: io.kestra.plugin.core.output.OutputValues
description: Summarize what was found in each system.
values:
database_rows: "{{ outputs.db_records.row.row_count }}"
tickets: "{{ outputs.tickets.response.hits.total.value }}"
ticket_ids: "{{ outputs.tickets.response.hits.hits | jq('[.[]._id]') | first |
join(', ') }}"
files: "{{ outputs.files.objects | length }}"
file_keys: "{{ outputs.files.objects | jq('[.[].key]') | first | join(', ') }}"
- id: by_type
type: io.kestra.plugin.core.flow.Switch
description: Export for access requests, erase for erasure requests.
value: "{{ inputs.request_type }}"
cases:
ACCESS:
- id: fetch_files
type: io.kestra.plugin.core.flow.If
description: Download the user's files only when there are any.
condition: "{{ outputs.files.objects | length > 0 and
outputs.subject.values.customer_id != '' }}"
then:
- id: download_files
type: io.kestra.plugin.aws.s3.Downloads
description: Copy the user's files into the execution. The originals are left
untouched.
endpointOverride: "{{ inputs.s3_endpoint }}"
forcePathStyle: true
region: us-east-1
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ inputs.bucket }}"
prefix: "customers/{{ outputs.subject.values.customer_id }}/"
action: NONE
compatibilityMode: true
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: personal_data_json
type: io.kestra.plugin.core.storage.Write
description: The database records as JSON.
extension: .json
content: "{{ outputs.db_records.row.export }}"
- id: tickets_json
type: io.kestra.plugin.core.storage.Write
description: The support tickets as JSON.
extension: .json
content: "{{ outputs.tickets.response.hits.hits | jq('[.[]._source]') | first |
toJson }}"
- id: export_bundle
type: io.kestra.plugin.compress.ArchiveCompress
description: Package everything into one ZIP to send to the data subject.
algorithm: ZIP
from: >-
{ {% for file in (outputs.download_files.outputFiles ?? {})
%}"files/{{ file.key | substringAfterLast('/') }}": "{{ file.value
}}", {% endfor %}"personal-data.json": "{{
outputs.personal_data_json.uri }}", "support-tickets.json": "{{
outputs.tickets_json.uri }}" }
- id: access_completed
type: io.kestra.plugin.core.kv.Set
description: Close the audit record.
key: "dsar-{{ inputs.request_id | slugify }}"
kvType: JSON
value: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "ACCESS", "status": "COMPLETED", "received_at": "{{ execution.startDate }}", "completed_at": "{{ now() }}", "execution_id": "{{ execution.id }}", "found": {"database_rows": {{ outputs.findings.values.database_rows }}, "tickets": {{ outputs.findings.values.tickets }}, "files": {{ outputs.findings.values.files }}}}
- id: notify_access
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Tell the privacy team the export is ready.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Access request {{ inputs.request_id }}* is ready: {{ outputs.findings.values.database_rows }} database row(s), {{ outputs.findings.values.tickets }} ticket(s) and {{ outputs.findings.values.files }} file(s) exported. Download the ZIP from the outputs of Kestra execution {{ execution.id }} and send it to the data subject.
ERASURE:
- id: erase_or_plan
type: io.kestra.plugin.core.flow.If
description: In dry run, only report the plan. Otherwise ask for approval,
erase, and verify.
condition: "{{ inputs.dry_run }}"
then:
- id: report_plan
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: List exactly what an erasure would change. Nothing is modified.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Erasure dry run for request {{ inputs.request_id }}*. Nothing was changed. A live run would:
- Postgres: delete {{ outputs.db_records.row.export | jq('.marketing_consents | length') | first }} marketing consent(s), anonymize the customer record, and clear the shipping address on {{ outputs.db_records.row.export | jq('.orders | length') | first }} order(s), keeping the order rows for financial records.
- OpenSearch: delete {{ outputs.findings.values.tickets }} ticket(s) from `{{ inputs.tickets_index }}`: {{ outputs.findings.values.ticket_ids }}
- Object storage: delete {{ outputs.findings.values.files }} file(s): {{ outputs.findings.values.file_keys }}
Re-run with dry_run set to false to erase after approval.
else:
- id: request_approval
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Ask the privacy team to approve the erasure.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Approval needed:* erase personal data for request {{ inputs.request_id }} ({{ outputs.findings.values.database_rows }} database row(s), {{ outputs.findings.values.tickets }} ticket(s), {{ outputs.findings.values.files }} file(s)). Resume Kestra execution {{ execution.id }} within 72 hours, filling in your name. If nobody does, nothing is erased.
- id: approval
type: io.kestra.plugin.core.flow.Pause
description: Wait for a named approver. After 72 hours the execution is
cancelled and nothing is erased.
pauseDuration: PT72H
behavior: CANCEL
onResume:
- id: approved
type: BOOL
description: Erase this person's data?
defaults: false
- id: approver
type: STRING
description: Your name, recorded in the proof of erasure.
- id: notes
type: STRING
description: Identity check performed, or reason for rejecting.
defaults: Identity verified
- id: decision
type: io.kestra.plugin.core.flow.If
description: Erase only when the approver said yes.
condition: "{{ outputs.approval.onResume.approved }}"
then:
- id: erase
type: io.kestra.plugin.core.flow.Parallel
description: Erase in every system at once.
tasks:
- id: erase_db
type: io.kestra.plugin.jdbc.postgresql.Queries
description: In one transaction, delete marketing consents, clear shipping
addresses on orders (the rows are kept for financial
records), and anonymize the customer record.
url: "{{ secret('PG_URL') }}"
username: "{{ secret('PG_USERNAME') }}"
password: "{{ secret('PG_PASSWORD') }}"
transaction: true
sql: |
DELETE FROM marketing_consents WHERE customer_id IN (SELECT id FROM customers WHERE lower(email) = lower(:email));
UPDATE orders SET shipping_address = NULL WHERE customer_id IN (SELECT id FROM customers WHERE lower(email) = lower(:email));
UPDATE customers SET email = 'erased-' || id || '@erased.invalid', full_name = NULL, phone = NULL, address = NULL, erased_at = now() WHERE lower(email) = lower(:email);
parameters:
email: "{{ inputs.subject_email }}"
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: erase_tickets
type: io.kestra.plugin.opensearch.Request
description: Delete the subject's support tickets and refresh the index so
verification sees the result.
connection:
hosts:
- "{{ secret('OPENSEARCH_URL') }}"
method: POST
endpoint: "/{{ inputs.tickets_index }}/_delete_by_query"
parameters:
refresh: "true"
body:
query:
term:
customer_email: "{{ inputs.subject_email | lower }}"
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: erase_files
type: io.kestra.plugin.aws.s3.DeleteList
description: Delete every file under the customer's prefix. Other customers'
files are untouched.
endpointOverride: "{{ inputs.s3_endpoint }}"
forcePathStyle: true
region: us-east-1
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ inputs.bucket }}"
prefix: "customers/{{ outputs.subject.values.customer_id != '' ?
outputs.subject.values.customer_id :
'no-customer-record' }}/"
errorOnEmpty: false
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
- id: verify
type: io.kestra.plugin.core.flow.Parallel
description: Search every system again to prove nothing is left.
tasks:
- id: verify_db
type: io.kestra.plugin.jdbc.postgresql.Query
description: Count anything still linked to the email or still holding personal
fields for this customer.
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
url: "{{ secret('PG_URL') }}"
username: "{{ secret('PG_USERNAME') }}"
password: "{{ secret('PG_PASSWORD') }}"
sql: |
SELECT
(SELECT count(*) FROM customers WHERE lower(email) = lower(:email))
+ (SELECT count(*) FROM customers WHERE id = CAST(:customer_id AS integer) AND (full_name IS NOT NULL OR phone IS NOT NULL OR address IS NOT NULL))
+ (SELECT count(*) FROM marketing_consents WHERE customer_id = CAST(:customer_id AS integer))
+ (SELECT count(*) FROM orders WHERE customer_id = CAST(:customer_id AS integer) AND shipping_address IS NOT NULL) AS residual
parameters:
email: "{{ inputs.subject_email }}"
customer_id: "{{ outputs.subject.values.customer_id != '' ?
outputs.subject.values.customer_id : -1 }}"
fetchType: FETCH_ONE
- id: verify_tickets
type: io.kestra.plugin.opensearch.Request
description: Count tickets still matching the email.
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
connection:
hosts:
- "{{ secret('OPENSEARCH_URL') }}"
method: POST
endpoint: "/{{ inputs.tickets_index }}/_count"
body:
query:
term:
customer_email: "{{ inputs.subject_email | lower }}"
- id: verify_files
type: io.kestra.plugin.aws.s3.List
description: Count files still under the customer's prefix.
retry:
type: exponential
interval: PT5S
maxInterval: PT1M
maxAttempts: 3
endpointOverride: "{{ inputs.s3_endpoint }}"
forcePathStyle: true
region: us-east-1
accessKeyId: "{{ secret('S3_ACCESS_KEY_ID') }}"
secretKeyId: "{{ secret('S3_SECRET_KEY_ID') }}"
bucket: "{{ inputs.bucket }}"
prefix: "customers/{{ outputs.subject.values.customer_id != '' ?
outputs.subject.values.customer_id :
'no-customer-record' }}/"
- id: fail_on_residual
type: io.kestra.plugin.core.execution.Fail
description: Refuse to mark the request complete if any personal data is left.
condition: "{{ outputs.verify_db.row.residual +
outputs.verify_tickets.response.count +
(outputs.verify_files.objects | length) > 0 }}"
errorMessage: "Residual personal data after erasure: {{
outputs.verify_db.row.residual }} database row(s), {{
outputs.verify_tickets.response.count }} ticket(s), {{
outputs.verify_files.objects | length }} file(s)."
- id: proof
type: io.kestra.plugin.core.storage.Write
description: Write the proof-of-erasure record for the audit trail.
extension: .json
content: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "ERASURE", "status": "COMPLETED", "received_at": "{{ execution.startDate }}", "erased_at": "{{ now() }}", "approver": {{ outputs.approval.onResume.approver | toJson }}, "notes": {{ outputs.approval.onResume.notes | toJson }}, "execution_id": "{{ execution.id }}", "erased": {"postgres": {"marketing_consents_deleted": {{ outputs.db_records.row.export | jq('.marketing_consents | length') | first }}, "orders_address_cleared": {{ outputs.db_records.row.export | jq('.orders | length') | first }}, "customer_anonymized": {{ outputs.subject.values.customer_id != '' }}}, "opensearch_tickets_deleted": {{ outputs.erase_tickets.response.deleted ?? 0 }}, "object_storage_files_deleted": {{ outputs.erase_files.count }}}, "verification": {"residual_database": {{ outputs.verify_db.row.residual }}, "residual_tickets": {{ outputs.verify_tickets.response.count }}, "residual_files": {{ outputs.verify_files.objects | length }}}}
- id: erasure_completed
type: io.kestra.plugin.core.kv.Set
description: Close the audit record with the proof.
key: "dsar-{{ inputs.request_id | slugify }}"
kvType: JSON
value: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "ERASURE", "status": "COMPLETED", "received_at": "{{ execution.startDate }}", "completed_at": "{{ now() }}", "approver": {{ outputs.approval.onResume.approver | toJson }}, "execution_id": "{{ execution.id }}", "proof": "{{ outputs.proof.uri }}"}
- id: notify_erasure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Confirm the erasure.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Erasure completed for request {{ inputs.request_id }}*, approved by {{ outputs.approval.onResume.approver }}. Verification found no personal data left. The proof of erasure is attached to Kestra execution {{ execution.id }}.
else:
- id: erasure_rejected
type: io.kestra.plugin.core.kv.Set
description: Record the rejection and the reason.
key: "dsar-{{ inputs.request_id | slugify }}"
kvType: JSON
value: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "ERASURE", "status": "REJECTED", "received_at": "{{ execution.startDate }}", "rejected_at": "{{ now() }}", "approver": {{ outputs.approval.onResume.approver | toJson }}, "notes": {{ outputs.approval.onResume.notes | toJson }}, "execution_id": "{{ execution.id }}"}
- id: notify_rejected
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Tell the team nothing was erased.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Erasure rejected for request {{ inputs.request_id }}* by {{ outputs.approval.onResume.approver }}: {{ outputs.approval.onResume.notes }}. Nothing was erased.
errors:
- id: mark_failed
type: io.kestra.plugin.core.kv.Set
description: Record the failure in the audit record.
key: "dsar-{{ inputs.request_id | slugify }}"
kvType: JSON
value: |
{"request_id": {{ inputs.request_id | toJson }}, "type": "{{ inputs.request_type }}", "status": "FAILED", "received_at": "{{ execution.startDate }}", "failed_at": "{{ now() }}", "execution_id": "{{ execution.id }}"}
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert the privacy team with the failed task, the error and a link
to the execution.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
retry:
type: constant
interval: PT10S
maxAttempts: 3
messageText: |
*Data subject request {{ inputs.request_id }} FAILED* in flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}.
Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'see the execution logs' }}
{{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}
The request is still open and due by the date in its audit record. Fix the cause and run the request again.
outputs:
- id: findings
type: JSON
displayName: Findings
description: What was found in each system.
value: "{{ outputs.findings.values }}"
- id: result_file
type: STRING
displayName: Result file
description: Internal storage URI of the export ZIP (access) or the proof of
erasure (erasure), when one was produced.
value: "{{ outputs.export_bundle.uri ?? outputs.proof.uri ?? 'none' }}"
