Verify and load source-backed company enrichment with Apify

id: source-backed-company-enrichment
namespace: company.team
description: |
  Enrich one company domain with website-derived metadata candidates, verify
  source provenance, and retain the Apify run and dataset IDs for audit.

inputs:
  - id: domain
    type: STRING
    displayName: Company domain
    description: One bare public company domain without a scheme, path, port,
      wildcard, or credentials.
    defaults: example.com

tasks:
  - id: enrich_company
    type: io.kestra.plugin.apify.actor.Run
    description: Run the fixed Brainiall Company Enrichment Actor once under
      one-result and US$0.02 total-charge caps.
    actorId: tFyauKW1J9LcscDT8
    apiToken: "{{ secret('APIFY_API_TOKEN') }}"
    maxItems: 1
    maxTotalChargeUsd: 0.02
    memory: MB_256
    requestTimeout: 120
    waitForFinish: 60
    input:
      domain: "{{ inputs.domain }}"
      integrationSource: kestra-c19

  - id: get_source_backed_result
    type: io.kestra.plugin.apify.dataset.Get
    description: Fetch the single default-dataset item created by the completed Actor run.
    apiToken: "{{ secret('APIFY_API_TOKEN') }}"
    datasetId: "{{ outputs.enrich_company.defaultDatasetId }}"

  - id: verify_source_backed_result
    type: io.kestra.plugin.core.execution.Assert
    description: Fail unless exactly one current website-scrape result with the
      Kestra route label is present.
    errorMessage: The enrichment result is empty, duplicated, lacks website-scrape
      provenance, or has an unexpected integration label.
    conditions:
      - "{{ (outputs.get_source_backed_result.dataset | length) == 1 }}"
      - "{{ outputs.get_source_backed_result.dataset[0].success == true }}"
      - "{{ outputs.get_source_backed_result.dataset[0].provenance.method ==
        'website_metadata_scrape' }}"
      - "{{ outputs.get_source_backed_result.dataset[0].integrationSource ==
        'kestra-c19' }}"

  - id: log_audit_receipt
    type: io.kestra.plugin.core.log.Log
    description: Log non-secret run and dataset identifiers so downstream operators
      can trace the result without logging the API token.
    message: "Validated one source-backed company metadata candidate. Actor run={{
      outputs.enrich_company.id }}, dataset={{
      outputs.enrich_company.defaultDatasetId }}, reported usage USD={{
      outputs.enrich_company.usageTotalUsd }}."

  - id: write_json
    type: io.kestra.plugin.core.storage.Write
    description: Persist the validated dataset array to internal storage as JSON.
    content: "{{ outputs.get_source_backed_result.dataset | json }}"
    extension: .json

  - id: reshape
    type: io.kestra.plugin.graalvm.python.FileTransform
    description: Reshape the raw candidate record into typed columns including
      provenance and source URL metadata.
    from: "{{ outputs.write_json.uri }}"
    script: |
      provenance = row.get("provenance") if isinstance(row.get("provenance"), dict) else {}
      row = {
          "domain": row.get("domain"),
          "website": row.get("website"),
          "name_candidate": row.get("nameCandidate"),
          "industry_candidate": row.get("industryCandidate"),
          "description": row.get("description"),
          "provenance_method": provenance.get("method"),
          "source_url": provenance.get("sourceUrl"),
          "provenance_caveat": provenance.get("caveat"),
          "observed_at": row.get("observedAt"),
          "integration_source": row.get("integrationSource"),
      }

  - id: to_csv
    type: io.kestra.plugin.serdes.csv.IonToCsv
    description: Serialize the reshaped enrichment record to CSV.
    from: "{{ outputs.reshape.uri }}"
    header: true

  - id: create_table
    type: io.kestra.plugin.jdbc.postgresql.Query
    description: Create the target Postgres table for company enrichment data if it
      does not exist.
    url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      CREATE TABLE IF NOT EXISTS public.source_backed_company_enrichment (
          domain              VARCHAR,
          website             VARCHAR,
          name_candidate      VARCHAR,
          industry_candidate  VARCHAR,
          description         TEXT,
          provenance_method   VARCHAR,
          source_url          VARCHAR,
          provenance_caveat   TEXT,
          observed_at         TIMESTAMPTZ,
          integration_source  VARCHAR,
          ingested_at         TIMESTAMPTZ DEFAULT now()
      );

  - id: load_data
    type: io.kestra.plugin.jdbc.postgresql.CopyIn
    description: Bulk-load the CSV record into the Postgres enrichment table.
    url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    format: CSV
    from: "{{ outputs.to_csv.uri }}"
    table: public.source_backed_company_enrichment
    header: true
    columns:
      - domain
      - website
      - name_candidate
      - industry_candidate
      - description
      - provenance_method
      - source_url
      - provenance_caveat
      - observed_at
      - integration_source