# streams/transaction.yaml
input:
  nats_jetstream:
    #urls: [ "nats-cluster:4222" ]
    urls: [ "nats://nats-cluster.infra.svc.cluster.local:4222" ]
    stream: "request_stream"
    subject: "input_transaction_logs"
    manage_stream: true # <--- Add this

# NEW: Global Pipeline for the input
pipeline:
  processors:
    - mapping: |
        # Set metadata globally so all outputs and templates can see it
        root = this
        meta conflict_col = this.data.order.id.or("unknown")
        
        # with thhe following assignment only streams can see the value but template not
        meta transaction_id = this.transaction_id.or("unknown")
        meta transaction_status = this.transaction_status
        root.timestamp = this.timestamp    
        meta order_id = this.data.order.id.or("unknown")
        # root.flat_status = this.data.order.status.lowercase()
        # root.order_customer_email = this.data.order.customer.email
        # root.order_customer_loyalty_tier = this.data.order.customer.loyalty_tier
        # root.metadata_ip_address = this.data.metadata.ip_address
        # root.metadata_region = this.data.metadata.region
        # root.metadata_user_agent = this.data.metadata.user_agent
        root.data.order.items = this.data.order.items
        # meta metadata_items = this.data.items.map_each(item -> item.sku)

    - log:
          level: INFO
          message: "New transaction: status ${! meta(\"transaction_status\") } - order ID: ${! meta(\"order_id\") } timestamp: ${! json(\"timestamp\")} timestamp2a ${! this.timestamp  }"
 
# output:
#   broker:
#     pattern: fan_out
#     outputs:
#       # --- Branch 1: Archive the raw JSON ---
#       - yugabyte_insert:
#           dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
#           table_name: "public.transaction"
#           #conflict_columns: "transaction_id"
#           #columns: ["transaction_id", "orders", "order_status"]
#           # if use this deactivate conflict_columns
#           columns: ["content_hash", "orders", "status","created_at"]
#           value_mapping: |
#             root = [
#               # if use meta define it first above
#               # meta("transaction_id"),
#               # meta("order_id"),
#               # meta("order_status")
#               this.transaction_id,
#               this.data.order.id,
#               this.transaction_status,
#               this.timestamp
#             ]           

#       # --- Branch 2: Extract nested GraphQL data into Facts ---
      
#       - yugabyte_insert:
#           dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
#           table_name: "public.order"
#           # conflict_columns: "content_hash"
#           columns: ["content_hash", "customer_email", "customer_loyalty_tier","status", "ip_address","region"]
#           # if use this deactivate conflict_columns
#           #columns: ["content_hash", "customer_email", "status"] 
#           value_mapping: |
#             root = [
#               # if use meta define it first above
#               # meta("order_id"), 
#               # meta("order_customer_email"),
#               # meta("order_status")
#               this.data.order.id,
#               this.data.order.customer.email,
#               this.data.order.customer.loyalty_tier,
#               this.data.order.status,
#               this.data.order.metadata.ip_address,
#               this.data.order.metadata.region            
#             ]

#       - yugabyte_insert:
#           dsn: "postgres://yugabyte@yugabytedb-3:5433/ecommerce?sslmode=disable"
#           table_name: "public.items"
#           columns: ["content_hash", "items"]
#           # No unarchive needed!
#           processors:
#             - mapping: |
#                 root = this
#                 # 1. Update dedupe key for this specific table
#                 meta conflict_col = meta("order_id") + "-item" 
                
#                 # 2. Stringify the items array for the JSONB column
#                 # root.items_blob = this.data.order.items.format_json()
          
#           value_mapping: |
#             root = [
#               meta("order_id"),
#               this.data.order.items.string()
#             ]
      
#       - kafka:
#           addresses: [ "redpanda:9092" ]
#           topic: "transactions_redpanda"
#           # By not adding a 'processors' block here, 
#           # it sends the current state of 'root' from the global pipeline.
#           # To ensure it is the ABSOLUTE original, we can force a mapping:
#           key : ${! meta("transaction_id")} 
#           processors:
#             - mapping: |
#                 root = content()
     
# 1. example below for using content_hash
# 2. change it accordingly to your defined conflict_columns
# 3. IMPORTANT: conflict_columns is to be defined as UNIQUE in sql
# CREATE TABLE public.transaction (
#     id BIGSERIAL, 
#     content_hash TEXT UNIQUE,
#     orders TEXT, 
#     order_status TEXT,
#     timestamp TIMESTAMPTZ,
#     PRIMARY KEY (id)
# );

# CREATE TABLE public.order (
#     id TEXT PRIMARY KEY,   -- Must be TEXT for "ORD-5521"
#     content_hash TEXT UNIQUE,
#     customer_email TEXT,
#     amount DECIMAL
# );