id: salesforce-postgres-sync
namespace: blueprints
tasks:
  - id: each
    type: io.kestra.plugin.core.flow.ForEach
    values: "{{ trigger.rows }}"
    tasks:
      - id: create_contacts_in_salesforce
        type: io.kestra.plugin.ee.salesforce.Create
        connection:
          username: "{{ secret('SALESFORCE_USERNAME') }}"
          password: "{{ secret('SALESFORCE_PASSWORD') }}"
          authEndpoint: "{{ secret('SALESFORCE_AUTH_ENDPOINT') }}"
        objectName: "Contact"
        records:
          - FirstName: "{{ json(taskrun.value).FirstName }}"
            LastName: "{{ json(taskrun.value).LastName }}"
            Email: "{{ json(taskrun.value).Email }}"

triggers:
  - id: postgres_trigger
    type: io.kestra.plugin.jdbc.postgresql.Trigger
    sql: |
      SELECT 
        first_name as "FirstName", 
        last_name as "LastName", 
        email as "Email"
      FROM customers
      WHERE updated_at > CURRENT_DATE - INTERVAL '1 day'
      AND (processed_at IS NULL OR processed_at < updated_at)
    interval: PT5M
    fetchType: FETCH

pluginDefaults:
  - type: io.kestra.plugin.jdbc.postgresql
    values:
      url: "{{ vars.postgres_url }}"
      username: "{{ vars.postgres_username }}"
      password: "{{ vars.postgres_password }}"