commit 21780412b803193cd4fb2219431ce87c6d9f6dda Author: pranajaya161 Date: Fri Aug 7 12:36:57 2026 +0700 Initial commit: extract omnix-broadcast as standalone repo omnix-sopiga Co-authored-by: Cursor diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..2006409 --- /dev/null +++ b/.env.example @@ -0,0 +1,15 @@ +DATABASE_DSN=postgres://user:pass@localhost:5432/gadai_mulia +DB_MAX_CONNS=25 +DB_MIN_CONNS=5 + +SOPIGA_BASE_URL=https://omnix-dev.promas.site +SOPIGA_TOKEN= + +WORKER_CHECK_INTERVAL_SEC=30 +WORKER_BATCH_SIZE=100 +WORKER_MAX_RETRIES=3 +WORKER_SYNC_INTERVAL_SEC=300 +WORKER_SYNC_BATCH_SIZE=100 + +WEBHOOK_PORT=8081 +WEBHOOK_SECRET= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..eeca4e9 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.go-cache/ +*.log +.env +.DS_Store + +# compiled binary from `go build` +/omnix-sopiga +/omnix-sopiga.exe diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml new file mode 100644 index 0000000..fece15d --- /dev/null +++ b/.gitlab-ci.yml @@ -0,0 +1,51 @@ +stages: + - test + - build + +variables: + GO_VERSION: "1.25" + # TODO: confirm final remote URL once go-dw-framework is pushed, and set up + # a deploy token / CI_JOB_TOKEN with read access if the repo is private. + FRAMEWORK_REPO_URL: "https://repository.promas.id/prana/go-dw-framework.git" + +# go.mod currently has `replace repository.promas.id/prana/go-dw-framework => ../go-dw-framework` +# for pre-release local development. CI clones it into that exact relative +# path so the replace directive resolves. Once go-dw-framework is tagged +# (e.g. v0.1.0), remove the replace line from go.mod, drop this clone step, +# and let `go mod download` fetch the tagged version normally. +.with-framework: + image: golang:${GO_VERSION}-alpine + before_script: + - apk add --no-cache git + - git clone --depth 1 "$FRAMEWORK_REPO_URL" ../go-dw-framework + cache: + key: go-mod-cache + paths: + - .go-cache/ + variables: + GOPATH: "$CI_PROJECT_DIR/.go-cache" + +test: + stage: test + extends: .with-framework + script: + - go vet ./... + - go test ./tests/... -v + +# NOTE: assumes the GitLab project path checks out into a folder literally +# named `omnix-sopiga` (matches CI_PROJECT_DIR's basename) so the Dockerfile's +# `COPY omnix-sopiga ./omnix-sopiga` line lines up. Adjust if your project slug differs. +build: + stage: build + image: docker:24 + services: + - docker:24-dind + before_script: + - apk add --no-cache git + - git clone --depth 1 "$FRAMEWORK_REPO_URL" ../go-dw-framework + script: + - docker build -f Dockerfile -t "$CI_REGISTRY_IMAGE:$CI_COMMIT_SHORT_SHA" .. + - docker login -u "$CI_REGISTRY_USER" -p "$CI_REGISTRY_PASSWORD" "$CI_REGISTRY" + - docker push "$CI_REGISTRY_IMAGE:$CI_COMMIT_SHORT_SHA" + rules: + - if: '$CI_COMMIT_BRANCH == "main"' diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..6b81a15 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,36 @@ +# Build context: PARENT directory of this repo, i.e. the folder that +# contains both `omnix-sopiga/` and `go-dw-framework/` as siblings: +# +# docker build -f omnix-sopiga/Dockerfile -t omnix-sopiga .. +# +# Why: go.mod still uses a local `replace` directive pointing at +# ../go-dw-framework (see go.mod) for pre-release development, since +# go-dw-framework isn't tagged/published yet. Once it is tagged (e.g. v0.1.0) +# and reachable from the Docker build environment: +# 1. go.mod: remove the `replace` line, keep `require ... v0.1.0` +# 2. This Dockerfile: build context becomes just this repo (`.`), and the +# COPY lines below collapse to `COPY . .` + +FROM golang:1.25-alpine AS builder + +RUN apk add --no-cache git + +WORKDIR /src +COPY go-dw-framework ./go-dw-framework +COPY omnix-sopiga ./omnix-sopiga + +WORKDIR /src/omnix-sopiga +RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/omnix-sopiga . + +FROM alpine:3.20 + +RUN apk add --no-cache ca-certificates tzdata && \ + adduser -D -u 10001 app +USER app + +WORKDIR /app +COPY --from=builder /out/omnix-sopiga . +COPY omnix-sopiga/migrations ./migrations + +EXPOSE 8081 +ENTRYPOINT ["./omnix-sopiga"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..ea6cb0b --- /dev/null +++ b/README.md @@ -0,0 +1,124 @@ +# omnix-sopiga + +Collection broadcast worker: mengirim invoice via WhatsApp (Sopiga Collar API) ke nasabah Gadai Mulia. + +> Repo standalone (bukan bagian monorepo). Dependency internal ada di +> [`go-dw-framework`](../go-dw-framework) — lihat `go.mod` untuk detail +> `replace` directive yang dipakai selama development lokal. + +## Architecture + +``` +Gadai Collection Service + → INSERT broadcast_staging (message_payload JSONB, status=pending) + → BroadcastWorker (poll every 30s) + → Query template_variable_mapping + → Build message dynamically + → POST Sopiga Collar API + → status=dispatched (sopiga_recipient_detail_id disimpan) + → DeliverySyncWorker (poll every 5 min) + → GET recipient detail dari Sopiga + → status=delivered (delivered_at diisi) atau status=failed +``` + +Tiga komponen jalan bersamaan (lihat `main.go`): +- **BroadcastWorker** (`worker/broadcast_worker.go`) — poll `pending`, dispatch ke Sopiga. +- **Webhook receiver** (`handler/webhook.go`, `POST /webhooks/sopiga/delivery-status`) — + jalur utama update status delivery. Begitu Omnix push callback, status langsung + diupdate tanpa perlu polling balik ke Sopiga sama sekali. +- **DeliverySyncWorker** (`worker/delivery_sync_worker.go`) — polling **fallback/backstop** + untuk record `dispatched` yang tidak kunjung dapat webhook (mis. delivery gagal + terkirim/network hiccup di sisi Omnix). Karena webhook jadi jalur utama, interval ini + bisa diperlonggar jauh lebih besar dari 5 menit tanpa menambah beban signifikan ke Omnix. + +> ⚠️ **Catatan:** payload webhook di `dto/webhook.go` dan skema signature (`X-Sopiga-Signature`, +> HMAC-SHA256) masih **asumsi kita sendiri** — belum dikonfirmasi tim Omnix, karena endpoint +> registrasi callback tidak ditemukan di dokumentasi resmi (`/docs?api-docs.yaml`) saat +> implementasi ini dibuat. Sebelum pakai di production: (1) konfirmasi ke tim Omnix apakah +> mereka support outgoing webhook untuk status collar recipient, (2) minta format payload & +> skema signature asli mereka, (3) sesuaikan `dto/webhook.go` dan `handler/webhook.go`, +> (4) daftarkan URL `/webhooks/sopiga/delivery-status` ke mereka. + +Template baru cukup didaftarkan lewat database (`sopiga_template_config` + +`template_variable_mapping` + `sopiga_collar_config`) — tanpa redeploy kode. +Lihat [docs/implementasi_guide.md](docs/implementasi_guide.md) dan +[docs/quick_start.md](docs/quick_start.md) untuk panduan lengkap. Detail +permintaan integrasi webhook ke tim Omnix ada di +[docs/webhook_integration_request.md](docs/webhook_integration_request.md). + +## Layers + +| Layer | Responsibility | +|---|---| +| `worker/` | Poller loop untuk dispatch + delivery sync fallback, fan-out ke concurrent processing | +| `handler/` | `WebhookHandler` — HTTP receiver untuk callback status delivery dari Omnix | +| `transformer/` | Entity ↔ Domain mapping, `BuildDynamicMessage` interpolasi template | +| `service/` | Business logic: fetch pending/dispatched, validate, dispatch, sync delivery (polling & webhook), retry, mark failed | +| `repository/` | Query `broadcast_staging` & `template_variable_mapping`, update status, lookup by `recipient_detail_id` | +| `client/` | HTTP client ke Sopiga Collar API (add-recipient + get recipient detail) | +| `domain/` | `Broadcast`, `TemplateVariable` | +| `entity/` | DB row mapping | +| `dto/` | Sopiga API request/response | +| `validator/` | Validasi payload sebelum dispatch | +| `migrations/` | Schema `collection_broadcasts` (tables, views, functions) | + +## Configuration (env vars) + +| Var | Default | +|---|---| +| `DATABASE_DSN` | — | +| `DB_MAX_CONNS` | 25 | +| `DB_MIN_CONNS` | 5 | +| `SOPIGA_BASE_URL` | `https://omnix.promas.site` | +| `SOPIGA_TOKEN` | — | +| `WORKER_CHECK_INTERVAL_SEC` | 30 | +| `WORKER_BATCH_SIZE` | 100 | +| `WORKER_MAX_RETRIES` | 3 | +| `WORKER_SYNC_INTERVAL_SEC` | 300 (fallback saja — perlonggar kalau webhook sudah aktif) | +| `WORKER_SYNC_BATCH_SIZE` | 100 | +| `WEBHOOK_PORT` | 8081 | +| `WEBHOOK_SECRET` | — (HMAC-SHA256 shared secret; kosongkan untuk skip verifikasi saat dev) | + +## Development setup + +Clone this repo and `go-dw-framework` as **siblings** (same parent folder) — +`go.mod` uses a local `replace` directive during pre-release development: + +``` +some-folder/ +├── go-dw-framework/ +└── omnix-sopiga/ ← you are here +``` + +```bash +git clone ../go-dw-framework +go build ./... +``` + +Once `go-dw-framework` is tagged (e.g. `v0.1.0`) and pushed to its remote, +switch `go.mod` to a plain `require repository.promas.id/prana/go-dw-framework v0.1.0` +(drop the `replace` line) and `go mod tidy`. + +## Run + +```bash +psql -d gadai_mulia -f migrations/001_create_collection_broadcasts.up.sql + +export DATABASE_DSN="postgres://user:pass@localhost:5432/gadai_mulia" +export SOPIGA_TOKEN="your_api_token" + +go run . +``` + +## Docker build + +```bash +# from the parent folder containing both omnix-sopiga/ and go-dw-framework/ +docker build -f omnix-sopiga/Dockerfile -t omnix-sopiga .. +``` + +## Test + +```bash +go test ./tests/... -v +``` diff --git a/client/sopiga.go b/client/sopiga.go new file mode 100644 index 0000000..fbf6dcb --- /dev/null +++ b/client/sopiga.go @@ -0,0 +1,107 @@ +package client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "time" + + "repository.promas.id/prana/omnix-sopiga/dto" +) + +type SopigaClient struct { + baseURL string + token string + httpClient *http.Client +} + +func NewSopigaClient(baseURL, token string) *SopigaClient { + return &SopigaClient{ + baseURL: baseURL, + token: token, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +func (c *SopigaClient) AddRecipient(ctx context.Context, req dto.CollarAddRecipientRequest) (*dto.CollarAPIResponse, error) { + body, err := json.Marshal(req) + if err != nil { + return nil, fmt.Errorf("marshal request: %w", err) + } + + httpReq, err := http.NewRequestWithContext( + ctx, + http.MethodPost, + fmt.Sprintf("%s/api/client/collar/add-recipient", c.baseURL), + bytes.NewReader(body), + ) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + + httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.token)) + httpReq.Header.Set("Content-Type", "application/json") + + httpResp, err := c.httpClient.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("http request: %w", err) + } + defer httpResp.Body.Close() + + bodyBytes, err := io.ReadAll(httpResp.Body) + if err != nil { + return nil, fmt.Errorf("read response body: %w", err) + } + + var resp dto.CollarAPIResponse + if err := json.Unmarshal(bodyBytes, &resp); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + + if httpResp.StatusCode != http.StatusAccepted && httpResp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("http status %d: %s", httpResp.StatusCode, resp.Message) + } + + return &resp, nil +} + +func (c *SopigaClient) GetRecipientDetail(ctx context.Context, recipientDetailID int64) (*dto.CollarRecipientDetailResponse, error) { + httpReq, err := http.NewRequestWithContext( + ctx, + http.MethodGet, + fmt.Sprintf("%s/api/client/collar/add-recipient/%d/detail", c.baseURL, recipientDetailID), + nil, + ) + if err != nil { + return nil, fmt.Errorf("create request: %w", err) + } + + httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.token)) + + httpResp, err := c.httpClient.Do(httpReq) + if err != nil { + return nil, fmt.Errorf("http request: %w", err) + } + defer httpResp.Body.Close() + + bodyBytes, err := io.ReadAll(httpResp.Body) + if err != nil { + return nil, fmt.Errorf("read response body: %w", err) + } + + var resp dto.CollarRecipientDetailResponse + if err := json.Unmarshal(bodyBytes, &resp); err != nil { + return nil, fmt.Errorf("unmarshal response: %w", err) + } + + if httpResp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("http status %d: %s", httpResp.StatusCode, resp.Message) + } + + return &resp, nil +} diff --git a/config.go b/config.go new file mode 100644 index 0000000..37f335c --- /dev/null +++ b/config.go @@ -0,0 +1,69 @@ +package main + +import ( + "time" + + fwconfig "repository.promas.id/prana/go-dw-framework/config" +) + +type Config struct { + DB DBConfig + Sopiga SopigaConfig + Worker WorkerConfig + Webhook WebhookConfig +} + +type DBConfig struct { + DSN string + MaxConns int32 + MinConns int32 + MaxConnLifetime time.Duration + MaxConnIdleTime time.Duration +} + +type SopigaConfig struct { + BaseURL string + Token string +} + +type WorkerConfig struct { + CheckInterval time.Duration + BatchSize int + MaxRetries int + RetryDelay time.Duration + SyncCheckInterval time.Duration + SyncBatchSize int +} + +type WebhookConfig struct { + Port string + Secret string +} + +func LoadConfig() *Config { + return &Config{ + DB: DBConfig{ + DSN: fwconfig.GetString("DATABASE_DSN", ""), + MaxConns: int32(fwconfig.GetInt("DB_MAX_CONNS", 25)), + MinConns: int32(fwconfig.GetInt("DB_MIN_CONNS", 5)), + MaxConnLifetime: 15 * time.Minute, + MaxConnIdleTime: 5 * time.Minute, + }, + Sopiga: SopigaConfig{ + BaseURL: fwconfig.GetString("SOPIGA_BASE_URL", "https://omnix.promas.site"), + Token: fwconfig.GetString("SOPIGA_TOKEN", ""), + }, + Worker: WorkerConfig{ + CheckInterval: time.Duration(fwconfig.GetInt("WORKER_CHECK_INTERVAL_SEC", 30)) * time.Second, + BatchSize: fwconfig.GetInt("WORKER_BATCH_SIZE", 100), + MaxRetries: fwconfig.GetInt("WORKER_MAX_RETRIES", 3), + RetryDelay: time.Second, + SyncCheckInterval: time.Duration(fwconfig.GetInt("WORKER_SYNC_INTERVAL_SEC", 300)) * time.Second, + SyncBatchSize: fwconfig.GetInt("WORKER_SYNC_BATCH_SIZE", 100), + }, + Webhook: WebhookConfig{ + Port: fwconfig.GetString("WEBHOOK_PORT", "8081"), + Secret: fwconfig.GetString("WEBHOOK_SECRET", ""), + }, + } +} diff --git a/docs/Collection_broadcast_worker_simplified.go b/docs/Collection_broadcast_worker_simplified.go new file mode 100644 index 0000000..963ab71 --- /dev/null +++ b/docs/Collection_broadcast_worker_simplified.go @@ -0,0 +1,452 @@ +package collector + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "io" + "log" + "net/http" + "strings" + "time" +) + +// ============ DATABASE MODELS ============ + +type BroadcastRecord struct { + ID int64 `db:"id"` + SopigaCollarID int `db:"sopiga_collar_id"` + SopigaTemplateID int `db:"sopiga_template_id"` + MessagePayload map[string]interface{} `db:"message_payload"` // JSONB + Status string `db:"status"` + ErrorMessage *string `db:"error_message"` + ErrorCount int `db:"error_count"` + SopigaRecipientDetailID *int64 `db:"sopiga_recipient_detail_id"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} + +type TemplateVariable struct { + VariableOrder int `db:"variable_order"` + SopigaVarName string `db:"sopiga_variable_name"` + VariableType string `db:"variable_type"` + DBFieldSource string `db:"db_field_source"` + IsRequired bool `db:"is_required"` +} + +// ============ SOPIGA API MODELS ============ + +type CollarAddRecipientRequest struct { + BroadcastID int `json:"broadcast_id"` + TemplateID int `json:"template_id"` + Details CollarDetails `json:"details"` + Attachment *CollarAttachment `json:"attachment,omitempty"` + Labels map[string]string `json:"labels,omitempty"` +} + +type CollarDetails struct { + Recipient string `json:"recipient"` + Message string `json:"message"` +} + +type CollarAttachment struct { + Type string `json:"type"` // document + Caption string `json:"caption"` + File string `json:"file"` // URL +} + +type CollarAPIResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + Data CollarResponseData `json:"data"` + Errors map[string][]string `json:"errors,omitempty"` +} + +type CollarResponseData struct { + BroadcastID int `json:"broadcast_id"` + RecipientDetailID int64 `json:"recipient_detail_id"` + Recipient string `json:"recipient"` + Status string `json:"status"` +} + +// ============ WORKER ============ + +type CollectionBroadcastWorker struct { + db *sql.DB + sopigaBaseURL string + sopigaToken string + httpClient *http.Client + checkInterval time.Duration + batchSize int + maxRetries int + logger *log.Logger +} + +func NewCollectionBroadcastWorker( + db *sql.DB, + sopigaBaseURL string, + sopigaToken string, + logger *log.Logger, +) *CollectionBroadcastWorker { + return &CollectionBroadcastWorker{ + db: db, + sopigaBaseURL: sopigaBaseURL, + sopigaToken: sopigaToken, + httpClient: &http.Client{ + Timeout: 30 * time.Second, + }, + checkInterval: 30 * time.Second, + batchSize: 100, + maxRetries: 3, + logger: logger, + } +} + +// Start — Main worker loop +func (w *CollectionBroadcastWorker) Start(ctx context.Context) { + w.logger.Println("Collection Broadcast Worker started") + + ticker := time.NewTicker(w.checkInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + w.logger.Println("Collection Broadcast Worker stopped") + return + case <-ticker.C: + w.processPendingRecords(ctx) + } + } +} + +// processPendingRecords — Fetch & process pending records +func (w *CollectionBroadcastWorker) processPendingRecords(ctx context.Context) { + records, err := w.getPendingRecords(ctx) + if err != nil { + w.logger.Printf("ERROR fetching pending records: %v", err) + return + } + + if len(records) == 0 { + return + } + + w.logger.Printf("Found %d pending records to process", len(records)) + + // Process concurrently (max 5) + sem := make(chan struct{}, 5) + for _, rec := range records { + sem <- struct{}{} + go func(r BroadcastRecord) { + defer func() { <-sem }() + w.processSingleRecord(ctx, r) + }(rec) + } + + // Wait + for i := 0; i < 5; i++ { + sem <- struct{}{} + } +} + +// getPendingRecords — Query broadcast_staging untuk pending records +// SIMPLIFIED: Hanya query dari broadcast_staging, semua data ada di message_payload +func (w *CollectionBroadcastWorker) getPendingRecords(ctx context.Context) ([]BroadcastRecord, error) { + query := ` + SELECT + id, sopiga_collar_id, sopiga_template_id, message_payload, + status, error_message, error_count, sopiga_recipient_detail_id, + created_at, updated_at + FROM collection_broadcasts.broadcast_staging + WHERE status = 'pending' + ORDER BY created_at ASC + LIMIT $1 + ` + + rows, err := w.db.QueryContext(ctx, query, w.batchSize) + if err != nil { + return nil, err + } + defer rows.Close() + + var records []BroadcastRecord + for rows.Next() { + var rec BroadcastRecord + var payloadJSON []byte + + err := rows.Scan( + &rec.ID, &rec.SopigaCollarID, &rec.SopigaTemplateID, &payloadJSON, + &rec.Status, &rec.ErrorMessage, &rec.ErrorCount, &rec.SopigaRecipientDetailID, + &rec.CreatedAt, &rec.UpdatedAt, + ) + if err != nil { + w.logger.Printf("ERROR scanning record: %v", err) + continue + } + + // Parse JSONB message_payload + err = json.Unmarshal(payloadJSON, &rec.MessagePayload) + if err != nil { + w.logger.Printf("ERROR parsing message_payload for record %d: %v", rec.ID, err) + continue + } + + records = append(records, rec) + } + + return records, rows.Err() +} + +// processSingleRecord — Call Sopiga API +func (w *CollectionBroadcastWorker) processSingleRecord(ctx context.Context, rec BroadcastRecord) { + // Build message dinamis dari template variable mapping + message, err := w.buildDynamicMessage(ctx, rec.SopigaTemplateID, rec.MessagePayload) + if err != nil { + w.logger.Printf("Record ID %d: ERROR building message: %v", rec.ID, err) + w.updateStatus(ctx, rec.ID, "failed", fmt.Sprintf("message build error: %v", err), nil) + return + } + + // Get recipient phone dari message_payload + recipientPhone, ok := rec.MessagePayload["nasabah_phone"].(string) + if !ok { + w.logger.Printf("Record ID %d: ERROR - nasabah_phone not in payload", rec.ID) + w.updateStatus(ctx, rec.ID, "failed", "nasabah_phone not in payload", nil) + return + } + + // Get invoice URL dari message_payload + invoiceURL, ok := rec.MessagePayload["invoice_pdf_url"].(string) + if !ok { + w.logger.Printf("Record ID %d: ERROR - invoice_pdf_url not in payload", rec.ID) + w.updateStatus(ctx, rec.ID, "failed", "invoice_pdf_url not in payload", nil) + return + } + + // Get nasabah_nama untuk caption + nasabahNama, _ := rec.MessagePayload["nasabah_nama"].(string) + + // Build attachment + attachment := &CollarAttachment{ + Type: "document", + Caption: fmt.Sprintf("Invoice - %s", nasabahNama), + File: invoiceURL, + } + + // Build labels + labels := map[string]string{ + "module": "gadai_collection", + } + + // Build request + collarReq := CollarAddRecipientRequest{ + BroadcastID: rec.SopigaCollarID, + TemplateID: rec.SopigaTemplateID, + Details: CollarDetails{ + Recipient: recipientPhone, + Message: message, + }, + Attachment: attachment, + Labels: labels, + } + + // Call API dengan retry + resp, err := w.callCollarAPIWithRetry(ctx, collarReq, 0) + if err != nil { + w.logger.Printf("Record ID %d: FAILED after %d retries: %v", rec.ID, w.maxRetries, err) + w.updateStatus(ctx, rec.ID, "failed", err.Error(), nil) + return + } + + // Success + w.logger.Printf("Record ID %d: DISPATCHED (recipient_detail_id: %d)", rec.ID, resp.Data.RecipientDetailID) + w.updateDispatchedStatus(ctx, rec.ID, resp.Data.RecipientDetailID) +} + +// buildDynamicMessage — Build message dari template variables mapping +// Query template_variable_mapping, interpolate sesuai urutan +func (w *CollectionBroadcastWorker) buildDynamicMessage( + ctx context.Context, + templateID int, + messagePayload map[string]interface{}, +) (string, error) { + // Query template variables + query := ` + SELECT variable_order, sopiga_variable_name, variable_type, db_field_source, is_required + FROM collection_broadcasts.template_variable_mapping + WHERE sopiga_template_id = $1 + ORDER BY variable_order ASC + ` + + rows, err := w.db.QueryContext(ctx, query, templateID) + if err != nil { + return "", fmt.Errorf("query template variables failed: %w", err) + } + defer rows.Close() + + var parts []string + for rows.Next() { + var tv TemplateVariable + if err := rows.Scan(&tv.VariableOrder, &tv.SopigaVarName, &tv.VariableType, &tv.DBFieldSource, &tv.IsRequired); err != nil { + return "", fmt.Errorf("scan template variable failed: %w", err) + } + + // Get value dari message_payload + value, exists := messagePayload[tv.DBFieldSource] + if !exists { + if tv.IsRequired { + return "", fmt.Errorf("missing required field: %s", tv.DBFieldSource) + } + value = "" + } + + // Format sesuai type + var formattedValue string + switch tv.VariableType { + case "string": + formattedValue = fmt.Sprintf("%v", value) + case "integer", "decimal": + formattedValue = fmt.Sprintf("%.0f", value) + case "date": + formattedValue = fmt.Sprintf("%v", value) + default: + formattedValue = fmt.Sprintf("%v", value) + } + + parts = append(parts, formattedValue) + } + + if err := rows.Err(); err != nil { + return "", err + } + + // Join dengan '#' + return strings.Join(parts, "#"), nil +} + +// callCollarAPIWithRetry — Call API dengan automatic retry +func (w *CollectionBroadcastWorker) callCollarAPIWithRetry( + ctx context.Context, + req CollarAddRecipientRequest, + attempt int, +) (*CollarAPIResponse, error) { + if attempt > w.maxRetries { + return nil, fmt.Errorf("max retries (%d) exceeded", w.maxRetries) + } + + resp, err := w.callCollarAPI(ctx, req) + if err != nil { + if attempt < w.maxRetries { + waitTime := time.Duration(1<= 0) +); + +-- Indexes untuk worker & monitoring queries +CREATE INDEX idx_broadcast_staging_status ON collection_broadcasts.broadcast_staging(status); +CREATE INDEX idx_broadcast_staging_created_at ON collection_broadcasts.broadcast_staging(created_at DESC); +CREATE INDEX idx_broadcast_staging_collar_id ON collection_broadcasts.broadcast_staging(sopiga_collar_id); +CREATE INDEX idx_broadcast_staging_template_id ON collection_broadcasts.broadcast_staging(sopiga_template_id); + +-- Composite index untuk worker (most frequent query) +CREATE INDEX idx_broadcast_staging_pending_query + ON collection_broadcasts.broadcast_staging(status, created_at ASC) + WHERE status = 'pending'; + +-- Composite index untuk retry queue +CREATE INDEX idx_broadcast_staging_retry_query + ON collection_broadcasts.broadcast_staging(status, error_count, created_at ASC) + WHERE status = 'retry_scheduled' AND error_count < 3; + +-- ============================================================================ +-- 3. AUDIT & LOGGING +-- ============================================================================ + +-- Status transition audit log +CREATE TABLE collection_broadcasts.broadcast_audit_log ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + old_status VARCHAR(50), + new_status VARCHAR(50) NOT NULL, + reason VARCHAR(500), + sopiga_response JSONB, + changed_by VARCHAR(100) DEFAULT 'system', + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_broadcast_audit_log_broadcast_id ON collection_broadcasts.broadcast_audit_log(broadcast_id); +CREATE INDEX idx_broadcast_audit_log_created_at ON collection_broadcasts.broadcast_audit_log(created_at DESC); + +-- Error tracking +CREATE TABLE collection_broadcasts.broadcast_error_log ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + error_type VARCHAR(100), -- network_error, api_error, validation_error + error_code VARCHAR(50), + error_message TEXT, + error_details JSONB, + attempt_number INT, + next_retry_at TIMESTAMP, + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_broadcast_error_log_broadcast_id ON collection_broadcasts.broadcast_error_log(broadcast_id); +CREATE INDEX idx_broadcast_error_log_error_type ON collection_broadcasts.broadcast_error_log(error_type); + +-- Sopiga delivery status sync +CREATE TABLE collection_broadcasts.sopiga_sync_job ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + sopiga_recipient_detail_id BIGINT, + last_synced_at TIMESTAMP, + last_status_from_sopiga VARCHAR(50), + sync_count INT DEFAULT 0, + next_sync_at TIMESTAMP, + completed_at TIMESTAMP, + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_sopiga_sync_job_broadcast_id ON collection_broadcasts.sopiga_sync_job(broadcast_id); +CREATE INDEX idx_sopiga_sync_job_next_sync_at ON collection_broadcasts.sopiga_sync_job(next_sync_at); + +-- ============================================================================ +-- 4. VIEWS +-- ============================================================================ + +-- Template variables dengan urutan (untuk Worker reference) +CREATE VIEW collection_broadcasts.v_template_variables_ordered AS +SELECT + tvm.sopiga_template_id, + stc.template_name, + tvm.variable_order, + tvm.sopiga_variable_name, + tvm.variable_type, + tvm.db_field_source, + tvm.is_required, + tvm.example_value +FROM collection_broadcasts.template_variable_mapping tvm +JOIN collection_broadcasts.sopiga_template_config stc ON tvm.sopiga_template_id = stc.sopiga_template_id +WHERE stc.active = TRUE +ORDER BY tvm.sopiga_template_id, tvm.variable_order; + +-- Collar summary +CREATE VIEW collection_broadcasts.v_collar_summary AS +SELECT + sc.sopiga_collar_id, + sc.broadcast_name, + COUNT(*) as total_records, + COUNT(CASE WHEN bs.status = 'pending' THEN 1 END) as pending, + COUNT(CASE WHEN bs.status = 'dispatched' THEN 1 END) as dispatched, + COUNT(CASE WHEN bs.status = 'delivered' THEN 1 END) as delivered, + COUNT(CASE WHEN bs.status = 'failed' THEN 1 END) as failed, + ROUND(100.0 * COUNT(CASE WHEN bs.status = 'delivered' THEN 1 END) / NULLIF(COUNT(*), 0), 2) as delivery_rate_percent +FROM collection_broadcasts.sopiga_collar_config sc +LEFT JOIN collection_broadcasts.broadcast_staging bs ON sc.sopiga_collar_id = bs.sopiga_collar_id +GROUP BY sc.sopiga_collar_id, sc.broadcast_name; + +-- Failed records (last 24h) +CREATE VIEW collection_broadcasts.v_failed_records_24h AS +SELECT + id, + sopiga_collar_id, + sopiga_template_id, + message_payload->>'nasabah_nama' as nasabah_nama, + message_payload->>'nasabah_phone' as nasabah_phone, + error_message, + error_count, + failed_at, + created_at +FROM collection_broadcasts.broadcast_staging +WHERE status = 'failed' AND created_at > NOW() - INTERVAL 1 DAY +ORDER BY failed_at DESC; + +-- Delivery rate (last 24h) +CREATE VIEW collection_broadcasts.v_delivery_rate_24h AS +SELECT + COUNT(*) as total, + COUNT(CASE WHEN status = 'delivered' THEN 1 END) as delivered, + COUNT(CASE WHEN status = 'failed' THEN 1 END) as failed, + COUNT(CASE WHEN status = 'dispatched' THEN 1 END) as in_progress, + ROUND(100.0 * COUNT(CASE WHEN status = 'delivered' THEN 1 END) / NULLIF(COUNT(*), 0), 2) as delivery_rate_percent +FROM collection_broadcasts.broadcast_staging +WHERE created_at > NOW() - INTERVAL 1 DAY; + +-- ============================================================================ +-- 5. HELPER FUNCTIONS +-- ============================================================================ + +-- Get template variables by template_id +CREATE OR REPLACE FUNCTION collection_broadcasts.get_template_variables( + p_sopiga_template_id INT +) +RETURNS TABLE( + variable_order INT, + sopiga_variable_name VARCHAR, + variable_type VARCHAR, + db_field_source VARCHAR, + is_required BOOLEAN +) AS $$ +BEGIN + RETURN QUERY + SELECT + tvm.variable_order, + tvm.sopiga_variable_name, + tvm.variable_type, + tvm.db_field_source, + tvm.is_required + FROM collection_broadcasts.template_variable_mapping tvm + WHERE tvm.sopiga_template_id = p_sopiga_template_id + ORDER BY tvm.variable_order ASC; +END; +$$ LANGUAGE plpgsql; + +-- Update status dengan audit log +CREATE OR REPLACE FUNCTION collection_broadcasts.update_broadcast_status( + p_broadcast_id BIGINT, + p_new_status VARCHAR, + p_error_message TEXT DEFAULT NULL, + p_sopiga_response JSONB DEFAULT NULL +) +RETURNS VOID AS $$ +DECLARE + v_old_status VARCHAR; +BEGIN + SELECT status INTO v_old_status + FROM collection_broadcasts.broadcast_staging + WHERE id = p_broadcast_id; + + UPDATE collection_broadcasts.broadcast_staging + SET + status = p_new_status, + error_message = p_error_message, + updated_at = NOW(), + dispatched_at = CASE WHEN p_new_status = 'dispatched' THEN NOW() ELSE dispatched_at END, + delivered_at = CASE WHEN p_new_status = 'delivered' THEN NOW() ELSE delivered_at END, + failed_at = CASE WHEN p_new_status = 'failed' THEN NOW() ELSE failed_at END + WHERE id = p_broadcast_id; + + INSERT INTO collection_broadcasts.broadcast_audit_log (broadcast_id, old_status, new_status, sopiga_response) + VALUES (p_broadcast_id, v_old_status, p_new_status, p_sopiga_response); +END; +$$ LANGUAGE plpgsql; + +-- ============================================================================ +-- 6. INITIALIZATION DATA +-- ============================================================================ + +-- Insert sample template +INSERT INTO collection_broadcasts.sopiga_template_config (template_name, sopiga_template_id, channel, template_type, description) +VALUES ('Collection Invoice May 2026', 2, 'waba', 'utility', 'Invoice bulanan untuk collection Mei 2026'); + +-- Insert template variables +INSERT INTO collection_broadcasts.template_variable_mapping (sopiga_template_id, variable_order, sopiga_variable_name, variable_type, db_field_source, is_required, example_value) +VALUES + (2, 1, 'Nama', 'string', 'nasabah_nama', TRUE, 'Budi Santoso'), + (2, 2, 'TotalTagihan', 'integer', 'nominal_tagihan', TRUE, '1500000'), + (2, 3, 'TanggalJatuhTempo', 'date', 'tanggal_tempo', TRUE, '2026-06-30'); + +-- Insert collar config +INSERT INTO collection_broadcasts.sopiga_collar_config (broadcast_name, sopiga_collar_id, sopiga_template_id, status, description) +VALUES ('Collection Invoices May 2026', 70, 2, 'open', 'Broadcast collar untuk collection invoice bulanan Mei 2026'); \ No newline at end of file diff --git a/docs/implementasi_guide.md b/docs/implementasi_guide.md new file mode 100644 index 0000000..8a55a7b --- /dev/null +++ b/docs/implementasi_guide.md @@ -0,0 +1,453 @@ +# Gadai Mulia Collection Broadcast Integration +## Complete Implementation Guide + +--- + +## 📋 Overview + +**Purpose:** Send collection invoices via WhatsApp (Sopiga Collar API) to Gadai Mulia customers. + +**Architecture:** +``` +Gadai Collection Service + → INSERT broadcast_staging (JSON payload) + → Go Worker (poll every 30s) + → Query template_variable_mapping + → Build message dynamically + → POST Sopiga Collar API + → WhatsApp delivery + → Update status (delivered/failed) +``` + +**Key Innovation:** Flexible template system via `template_variable_mapping` table — NO code redeploy for new templates. + +--- + +## 🗄️ Database Schema (Simplified Denormalized) + +### Master Configuration (Read-only) + +**Table: sopiga_template_config** +- Stores Sopiga template references (pre-created in Sopiga) +- Fields: `id`, `sopiga_template_id`, `template_name`, `channel` (waba), `template_type`, `description`, `active` +- Pre-populate: One row per template in Sopiga + +**Table: template_variable_mapping** +- Maps Sopiga template variables → message_payload keys +- **CRITICAL:** `variable_order` determines message interpolation sequence +- Fields: `id`, `sopiga_template_id`, `variable_order`, `sopiga_variable_name`, `variable_type` (string/integer/date/decimal), `db_field_source`, `is_required`, `example_value` +- Example: + ``` + sopiga_template_id=2, order=1, name='Nama', field='nasabah_nama' + sopiga_template_id=2, order=2, name='TotalTagihan', field='nominal_tagihan' + sopiga_template_id=2, order=3, name='TanggalJatuhTempo', field='tanggal_tempo' + ``` + +**Table: sopiga_collar_config** +- Stores Sopiga collar references (pre-created in Sopiga) +- Fields: `id`, `sopiga_collar_id`, `broadcast_name`, `sopiga_template_id` (FK), `status` (open/closed), `description` +- Pre-populate: One row per collar in Sopiga + +### Processing (Core) + +**Table: broadcast_staging** ← Main table +- DENORMALIZED: All data in `message_payload` (JSONB) +- NO FK to nasabah or gadai_contract +- Fields: + - `id` (PK) + - `sopiga_collar_id` (FK) + - `sopiga_template_id` (FK) + - `message_payload` (JSONB) ← **Single source of truth** + - `status` (pending → dispatched → delivered/failed) + - `error_message`, `error_count` + - `sopiga_recipient_detail_id` (response from Sopiga) + - `created_at`, `updated_at`, `dispatched_at`, `delivered_at`, `failed_at` + +**Example message_payload:** +```json +{ + "nasabah_nama": "Budi Santoso", + "nasabah_phone": "6281234567890", + "nominal_tagihan": 1500000, + "tanggal_tempo": "2026-06-30", + "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf", + "contract_no": "GAD-2026-001" +} +``` + +### Audit & Tracking + +**Table: broadcast_audit_log** +- Tracks status transitions +- Fields: `id`, `broadcast_id` (FK), `old_status`, `new_status`, `reason`, `sopiga_response` (JSONB), `changed_by`, `created_at` + +**Table: broadcast_error_log** +- Detailed error tracking +- Fields: `id`, `broadcast_id` (FK), `error_type`, `error_code`, `error_message`, `error_details` (JSONB), `attempt_number`, `next_retry_at` + +**Table: sopiga_sync_job** +- Tracks delivery status sync from Sopiga +- Fields: `id`, `broadcast_id` (FK), `sopiga_recipient_detail_id`, `last_synced_at`, `last_status_from_sopiga`, `sync_count`, `next_sync_at` + +--- + +## 🔧 Setup Process + +### Step 1: Setup in Sopiga (Manual) + +1. **Create template** in Sopiga: + ``` + POST https://omnix.promas.site/api/client/template + { + "template_name": "Collection Invoice", + "channel": "waba", + "variables": [ + {"name": "Nama", "type": "string"}, + {"name": "TotalTagihan", "type": "integer"}, + {"name": "TanggalJatuhTempo", "type": "date"} + ] + } + → Response: template_id = 2 + ``` + +2. **Create collar** in Sopiga: + ``` + POST https://omnix.promas.site/api/client/collar + { + "judul_broadcast": "Invoice May 2026", + "template_id": 2 + } + → Response: collar_id = 70 + ``` + +### Step 2: Register in Database (5 minutes) + +**2.1 Insert template config:** +```sql +INSERT INTO collection_broadcasts.sopiga_template_config + (template_name, sopiga_template_id, channel, template_type, description) +VALUES + ('Collection Invoice May 2026', 2, 'waba', 'utility', 'Invoice bulanan'); +``` + +**2.2 Insert variable mappings (ORDER IS CRITICAL!):** +```sql +INSERT INTO collection_broadcasts.template_variable_mapping + (sopiga_template_id, variable_order, sopiga_variable_name, variable_type, db_field_source, is_required, example_value) +VALUES + (2, 1, 'Nama', 'string', 'nasabah_nama', TRUE, 'Budi Santoso'), + (2, 2, 'TotalTagihan', 'integer', 'nominal_tagihan', TRUE, '1500000'), + (2, 3, 'TanggalJatuhTempo', 'date', 'tanggal_tempo', TRUE, '2026-06-30'); +``` + +**2.3 Insert collar config:** +```sql +INSERT INTO collection_broadcasts.sopiga_collar_config + (broadcast_name, sopiga_collar_id, sopiga_template_id, status) +VALUES + ('Collection Invoices May 2026', 70, 2, 'open'); +``` + +### Step 3: Deploy Application + +Deploy Go worker with: +- Database connection string +- Sopiga base URL: `https://omnix.promas.site` +- Sopiga API token + +Worker runs continuously: +- **Broadcast dispatch:** Poll every 30s, process pending records +- **Status sync:** Poll every 5 min, sync delivery status from Sopiga + +--- + +## 🔄 How It Works + +### Runtime Flow + +**Gadai Collection Service (sends data):** +```go +// When collection reminder needed: +payload := map[string]interface{}{ + "nasabah_nama": "Budi Santoso", + "nasabah_phone": "6281234567890", + "nominal_tagihan": 1500000.0, + "tanggal_tempo": "2026-06-30", + "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf", + "contract_no": "GAD-2026-001", +} + +// Insert to broadcast_staging +db.Exec(` + INSERT INTO collection_broadcasts.broadcast_staging + (sopiga_collar_id, sopiga_template_id, message_payload, status) + VALUES ($1, $2, $3, 'pending') +`, 70, 2, payload) +``` + +**Go Worker (processes data - automatic):** +``` +1. Poll broadcast_staging WHERE status='pending' LIMIT 100 +2. For each record: + a. Get message_payload from database (already has all fields) + b. Query template_variable_mapping WHERE sopiga_template_id=2 ORDER BY variable_order + c. Build message: + - For each variable in order (1,2,3...): + - Get value from payload[db_field_source] + - Format by type (string/integer/date) + - Append to parts[] + - message = parts.join("#") + - Result: "Budi Santoso#1500000#2026-06-30" + + d. Call Sopiga API: + POST /api/client/collar/add-recipient + { + "broadcast_id": 70, + "template_id": 2, + "details": { + "recipient": "6281234567890", + "message": "Budi Santoso#1500000#2026-06-30" + }, + "attachment": { + "type": "document", + "file": "https://storage.gadai.com/invoices/inv-001.pdf" + } + } + + e. Response (HTTP 202 Accepted): + { + "success": true, + "data": { + "recipient_detail_id": 512, + "status": "pending" + } + } + + f. Update broadcast_staging: + UPDATE status='dispatched', sopiga_recipient_detail_id=512 + +3. Status sync worker (every 5 min): + - For dispatched records: + - GET /api/client/collar/add-recipient/{recipient_detail_id}/detail + - Update status: delivered OR failed +``` + +--- + +## 📊 Adding New Template (Zero Code Change) + +### Scenario: Template now has 4 variables instead of 3 + +**Old:** +``` +1. Nama +2. TotalTagihan +3. TanggalJatuhTempo +``` + +**New:** +``` +1. Nama +2. NoKontrak ← NEW! +3. TotalTagihan +4. TanggalJatuhTempo +``` + +### Setup (5 minutes, Database only) + +1. Create template in Sopiga → template_id = 3 +2. Create collar in Sopiga → collar_id = 71 + +3. Register in database: +```sql +-- Template +INSERT INTO sopiga_template_config (template_name, sopiga_template_id, ...) +VALUES ('Collection Invoice Extended', 3, ...); + +-- Variables (URUTAN PENTING!) +INSERT INTO template_variable_mapping + (sopiga_template_id, variable_order, sopiga_variable_name, variable_type, db_field_source, ...) +VALUES + (3, 1, 'Nama', 'string', 'nasabah_nama', ...), + (3, 2, 'NoKontrak', 'string', 'contract_no', ...), ← NEW! + (3, 3, 'TotalTagihan', 'integer', 'nominal_tagihan', ...), + (3, 4, 'TanggalJatuhTempo', 'date', 'tanggal_tempo', ...); + +-- Collar +INSERT INTO sopiga_collar_config (broadcast_name, sopiga_collar_id, sopiga_template_id, ...) +VALUES ('Collection Invoices June 2026', 71, 3, ...); +``` + +**Go code?** ✅ ZERO changes — `buildDynamicMessage()` queries mapping every time. + +--- + +## 🚀 Gadai Service Integration + +**How Gadai Collection Service calls this:** + +```go +package gadai + +type CollectionBroadcastService struct { + db *sql.DB +} + +func (s *CollectionBroadcastService) SendInvoiceReminder( + ctx context.Context, + nasabahID, contractNo string, + nominalTagihan float64, + dueDate time.Time, +) error { + // Get invoice URL (from Gadai storage or API) + invoiceURL := fmt.Sprintf("https://storage.gadai.com/invoices/%s.pdf", contractNo) + + // Build payload (flexible - can have extra fields) + payload := map[string]interface{}{ + "nasabah_nama": "Budi Santoso", + "nasabah_phone": "6281234567890", + "nominal_tagihan": nominalTagihan, + "tanggal_tempo": dueDate.Format("2006-01-02"), + "invoice_pdf_url": invoiceURL, + "contract_no": contractNo, + "cif": "12345", // Extra field - ignored by worker if not in mapping + } + + // Marshal to JSONB + payloadJSON, _ := json.Marshal(payload) + + // Insert to broadcast_staging + // Worker will automatically pick it up (every 30s poll) + query := ` + INSERT INTO collection_broadcasts.broadcast_staging + (sopiga_collar_id, sopiga_template_id, message_payload, status) + VALUES ($1, $2, $3, 'pending') + ` + + _, err := s.db.ExecContext(ctx, query, + 70, // sopiga_collar_id (hardcoded or from config) + 2, // sopiga_template_id (hardcoded or from config) + payloadJSON, + ) + + return err +} +``` + +--- + +## 📈 Monitoring & Reporting + +### Views Available + +**v_template_variables_ordered** +```sql +SELECT * FROM collection_broadcasts.v_template_variables_ordered; +-- Shows all template variables with proper ordering +``` + +**v_collar_summary** +```sql +SELECT * FROM collection_broadcasts.v_collar_summary; +-- Shows delivery rate per collar +``` + +**v_delivery_rate_24h** +```sql +SELECT * FROM collection_broadcasts.v_delivery_rate_24h; +-- Shows delivery rate for last 24 hours +``` + +**v_failed_records_24h** +```sql +SELECT * FROM collection_broadcasts.v_failed_records_24h; +-- Shows failed records with error details +``` + +--- + +## 🛠️ Troubleshooting + +### Issue: Message format wrong + +**Cause:** `variable_order` in mapping doesn't match Sopiga template order + +**Fix:** +```sql +-- Check current mapping +SELECT variable_order, sopiga_variable_name FROM template_variable_mapping +WHERE sopiga_template_id=2 ORDER BY variable_order; + +-- If order wrong, delete and re-insert correctly +DELETE FROM template_variable_mapping WHERE sopiga_template_id=2; +-- Re-insert with correct order +``` + +### Issue: Missing field in message_payload + +**Cause:** Gadai service didn't include field in payload, but it's marked required in mapping + +**Fix:** +```sql +-- Either make it optional +UPDATE template_variable_mapping SET is_required=FALSE +WHERE sopiga_variable_name='NoKontrak'; + +-- Or ensure Gadai service includes it +``` + +### Issue: Delivery status stuck at 'dispatched' + +**Cause:** Status sync worker not running or Sopiga API unreachable + +**Fix:** +```sql +-- Check sync jobs +SELECT * FROM sopiga_sync_job WHERE next_sync_at < NOW(); + +-- Manually trigger status check +SELECT collection_broadcasts.get_template_variables(2); +``` + +--- + +## 🔐 Security Considerations + +1. **Sopiga API Token:** Store in environment variable, never hardcode +2. **Database:** Restrict access to `collection_broadcasts` schema +3. **Phone Numbers:** Already PII, treat as sensitive +4. **PDFs:** Ensure URLs are time-limited or access-controlled + +--- + +## 📋 Checklist: New Template + +- [ ] Create template in Sopiga, note `template_id` +- [ ] Create collar in Sopiga, note `collar_id` +- [ ] Insert `sopiga_template_config` row +- [ ] Insert `template_variable_mapping` rows (check order!) +- [ ] Insert `sopiga_collar_config` row +- [ ] Verify mapping: `SELECT * FROM v_template_variables_ordered WHERE sopiga_template_id=X` +- [ ] Test with sample record in `broadcast_staging` +- [ ] Monitor first 10 deliveries in logs +- [ ] Confirm in `v_collar_summary` delivery rate + +--- + +## 📞 Support + +**Schema files:** +- `gadai_collection_broadcast_simplified_schema.sql` — Complete DDL + +**Application files:** +- `collection_broadcast_worker_simplified.go` — Go worker implementation + +**Deployment:** +- Requires: PostgreSQL 11+, Go 1.16+ +- Connects to: Sopiga API at `https://omnix.promas.site` + +--- + +**Version:** 1.0 +**Last Updated:** 2026-08-03 +**Status:** Ready for deployment ✅ \ No newline at end of file diff --git a/docs/quick_start.md b/docs/quick_start.md new file mode 100644 index 0000000..e2b83cd --- /dev/null +++ b/docs/quick_start.md @@ -0,0 +1,319 @@ +# Quick Start Guide +## Gadai Mulia Collection Broadcast Integration + +--- + +## ⚡ 5-Minute Setup + +### Prerequisites +- PostgreSQL database access +- Sopiga API credentials (token) +- Go 1.16+ (for worker deployment) + +### Step 1: Create Database Schema (2 minutes) + +```bash +# Run SQL schema +psql -U postgres -h localhost -d gadai_mulia < gadai_collection_broadcast_simplified_schema.sql +``` + +Verify schema created: +```sql +\dt collection_broadcasts.* +-- Should show: sopiga_template_config, template_variable_mapping, sopiga_collar_config, broadcast_staging, etc +``` + +### Step 2: Setup Sopiga Template (2 minutes) + +**In Sopiga UI or API:** + +```bash +# Create template +curl -X POST https://omnix.promas.site/api/client/template \ + -H "Authorization: Bearer YOUR_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "template_name": "Collection Invoice", + "channel": "waba", + "variables": [ + {"name": "Nama", "type": "string"}, + {"name": "TotalTagihan", "type": "integer"}, + {"name": "TanggalJatuhTempo", "type": "date"} + ] + }' +# Response: {"data": {"id": 2}} + +# Create collar +curl -X POST https://omnix.promas.site/api/client/collar \ + -H "Authorization: Bearer YOUR_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{ + "judul_broadcast": "Invoice May 2026", + "template_id": 2 + }' +# Response: {"data": {"id": 70}} +``` + +Note down: +- `template_id = 2` +- `collar_id = 70` + +### Step 3: Register in Database (1 minute) + +**Connect to PostgreSQL:** +```bash +psql -U postgres -h localhost -d gadai_mulia +``` + +**Insert template:** +```sql +INSERT INTO collection_broadcasts.sopiga_template_config + (template_name, sopiga_template_id, channel, template_type, description) +VALUES + ('Collection Invoice May 2026', 2, 'waba', 'utility', 'Invoice bulanan untuk collection'); +``` + +**Insert variable mappings (URUTAN PENTING!):** +```sql +INSERT INTO collection_broadcasts.template_variable_mapping + (sopiga_template_id, variable_order, sopiga_variable_name, variable_type, db_field_source, is_required, example_value) +VALUES + (2, 1, 'Nama', 'string', 'nasabah_nama', TRUE, 'Budi Santoso'), + (2, 2, 'TotalTagihan', 'integer', 'nominal_tagihan', TRUE, '1500000'), + (2, 3, 'TanggalJatuhTempo', 'date', 'tanggal_tempo', TRUE, '2026-06-30'); +``` + +**Insert collar config:** +```sql +INSERT INTO collection_broadcasts.sopiga_collar_config + (broadcast_name, sopiga_collar_id, sopiga_template_id, status, description) +VALUES + ('Collection Invoices May 2026', 70, 2, 'open', 'Broadcast collar untuk collection invoice bulanan Mei 2026'); +``` + +**Verify setup:** +```sql +SELECT * FROM collection_broadcasts.v_template_variables_ordered +WHERE sopiga_template_id = 2; +-- Should show 3 rows with order 1, 2, 3 +``` + +### Step 4: Deploy Worker (1 minute) + +```bash +# Build +go build -o collection_broadcast_worker collection_broadcast_worker_simplified.go + +# Run with environment variables +export DB_HOST=localhost +export DB_PORT=5432 +export DB_NAME=gadai_mulia +export DB_USER=postgres +export DB_PASSWORD=xxx +export SOPIGA_BASE_URL=https://omnix.promas.site +export SOPIGA_TOKEN=your_api_token + +./collection_broadcast_worker +# Output: Collection Broadcast Worker started +``` + +--- + +## 🧪 Test It + +### Insert Test Record + +```sql +-- Insert to broadcast_staging +INSERT INTO collection_broadcasts.broadcast_staging + (sopiga_collar_id, sopiga_template_id, message_payload, status) +VALUES + ( + 70, + 2, + '{"nasabah_nama": "Budi Santoso", "nasabah_phone": "6281234567890", "nominal_tagihan": 1500000, "tanggal_tempo": "2026-06-30", "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf"}', + 'pending' + ); + +-- Check it's pending +SELECT id, status, created_at FROM collection_broadcasts.broadcast_staging WHERE status='pending'; +``` + +### Wait 30 seconds + +Worker polls every 30 seconds. Check status: + +```sql +SELECT id, status, sopiga_recipient_detail_id, dispatched_at, error_message +FROM collection_broadcasts.broadcast_staging +WHERE id = 1; + +-- Should show: status='dispatched', sopiga_recipient_detail_id=512 +``` + +### Check WhatsApp Delivery + +After 5 minutes, check sync job: + +```sql +SELECT id, status, last_status_from_sopiga, delivered_at +FROM collection_broadcasts.broadcast_staging +WHERE id = 1; + +-- Should show: status='delivered', delivered_at= +``` + +--- + +## 📊 Monitor + +### Check Delivery Rate +```sql +SELECT * FROM collection_broadcasts.v_delivery_rate_24h; +-- delivery_rate_percent should be > 95% +``` + +### Check Failed Records +```sql +SELECT * FROM collection_broadcasts.v_failed_records_24h; +-- Should be empty or < 5% +``` + +### Check Collar Summary +```sql +SELECT * FROM collection_broadcasts.v_collar_summary; +-- See totals per collar, delivery rates +``` + +--- + +## ➕ Add New Template (5 minutes) + +When Sopiga template changes (e.g., add `NoKontrak` field): + +### 1. Create in Sopiga +```bash +# template_id = 3, collar_id = 71 +``` + +### 2. Register in Database +```sql +-- Template +INSERT INTO sopiga_template_config (template_name, sopiga_template_id, ...) +VALUES ('Collection Invoice Extended', 3, ...); + +-- Variables (URUTAN PENTING!) +INSERT INTO template_variable_mapping + (sopiga_template_id, variable_order, sopiga_variable_name, variable_type, db_field_source, is_required) +VALUES + (3, 1, 'Nama', 'string', 'nasabah_nama', TRUE), + (3, 2, 'NoKontrak', 'string', 'contract_no', TRUE), ← NEW + (3, 3, 'TotalTagihan', 'integer', 'nominal_tagihan', TRUE), + (3, 4, 'TanggalJatuhTempo', 'date', 'tanggal_tempo', TRUE); + +-- Collar +INSERT INTO sopiga_collar_config (broadcast_name, sopiga_collar_id, sopiga_template_id, status) +VALUES ('Collection Invoices June 2026', 71, 3, 'open'); +``` + +### 3. NO code redeploy needed ✅ + +Worker automatically picks up new template on next poll. + +--- + +## 🚀 Gadai Service Integration + +**How Gadai calls this:** + +```go +import "database/sql" +import "encoding/json" + +db, _ := sql.Open("postgres", "postgres://user:pass@localhost/gadai_mulia") + +payload := map[string]interface{}{ + "nasabah_nama": "Budi Santoso", + "nasabah_phone": "6281234567890", + "nominal_tagihan": 1500000.0, + "tanggal_tempo": "2026-06-30", + "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf", + "contract_no": "GAD-2026-001", +} + +payloadJSON, _ := json.Marshal(payload) + +db.Exec(` + INSERT INTO collection_broadcasts.broadcast_staging + (sopiga_collar_id, sopiga_template_id, message_payload, status) + VALUES ($1, $2, $3, 'pending') +`, 70, 2, payloadJSON) + +// Done! Worker picks it up automatically +``` + +--- + +## 🐛 Troubleshooting + +### "Message format wrong" +→ Check template variable order matches Sopiga: +```sql +SELECT variable_order, sopiga_variable_name FROM template_variable_mapping +WHERE sopiga_template_id=2 ORDER BY variable_order; +``` + +### "Record stuck in pending" +→ Check worker logs: +```bash +# Look for errors in stdout/stderr +# Check DB connection +psql -U postgres -h localhost -d gadai_mulia -c "SELECT 1;" +``` + +### "Sopiga API errors" +→ Check token and URL: +```bash +curl -X GET https://omnix.promas.site/api/client/collar/list \ + -H "Authorization: Bearer YOUR_TOKEN" +``` + +--- + +## ✅ Checklist + +- [ ] Schema created +- [ ] Sopiga template created (note template_id) +- [ ] Sopiga collar created (note collar_id) +- [ ] Database populated with template_id, collar_id +- [ ] Variable mappings inserted (check order!) +- [ ] Worker deployed and running +- [ ] Test record inserted +- [ ] Record moved to 'dispatched' after 30s +- [ ] Record moved to 'delivered' after 5 min +- [ ] WhatsApp message received on test phone + +--- + +## 📞 Quick Reference + +**Database:** +- Schema: `collection_broadcasts` +- Main table: `broadcast_staging` +- Config tables: `sopiga_template_config`, `template_variable_mapping`, `sopiga_collar_config` + +**Sopiga URLs:** +- Base: `https://omnix.promas.site` +- Create template: POST `/api/client/template` +- Create collar: POST `/api/client/collar` +- Add recipient: POST `/api/client/collar/add-recipient` +- Get recipient detail: GET `/api/client/collar/add-recipient/{id}/detail` + +**Worker:** +- Poll interval: 30 seconds (pending records) +- Status sync: 5 minutes (delivery status) +- Max retries: 3 (exponential backoff: 1s, 2s, 4s) + +--- + +**Ready to go! 🚀** \ No newline at end of file diff --git a/docs/webhook_integration_request.md b/docs/webhook_integration_request.md new file mode 100644 index 0000000..e80486e --- /dev/null +++ b/docs/webhook_integration_request.md @@ -0,0 +1,89 @@ +# Permintaan Integrasi Webhook — Collection Broadcast + +**Ke:** Tim Omnix +**Dari:** Tim Collection Broadcast (Gadai Mulia) +**Tujuan:** Kami butuh notifikasi real-time saat status pengiriman WhatsApp +(sent/delivered/read/failed) berubah, supaya tidak perlu polling +`GET /api/client/collar/add-recipient/{id}/detail` berulang-ulang ke sistem +Omnix untuk setiap recipient. + +--- + +## 1. Yang kami minta dari tim Omnix + +1. **Konfirmasi ketersediaan fitur** — apakah Omnix/Sopiga sudah punya (atau bisa + dibuatkan) mekanisme outgoing webhook saat status recipient di collar berubah? + (Kami cek dokumentasi publik di `/docs?api-docs.yaml` dan tidak menemukan + endpoint registrasi webhook untuk collar/recipient — hanya ada `webhook_url` + di level `WhatsAppSession`, yang tampaknya untuk keperluan lain.) +2. **Endpoint/cara registrasi** URL callback kami ke sistem Omnix (dashboard, + API, atau config manual). +3. **Shared secret** untuk signing payload (lihat §3) — dikirim lewat kanal aman + (bukan email biasa), atau kalau Omnix punya skema signature sendiri (mis. HMAC + dengan public key, atau format berbeda), kasih tahu kami spesifikasinya — + kami sesuaikan. +4. **Kapan callback dikirim** — idealnya setiap kali status berubah + (`sent` → `delivered` → `read`, atau → `failed`), bukan cuma sekali di akhir. +5. **Retry policy di sisi Omnix** — kalau endpoint kami down/timeout, apakah + Omnix retry otomatis? Berapa kali, dengan interval berapa? + +## 2. Endpoint & secret yang kami berikan ke Omnix + +``` +URL : https:///webhooks/sopiga/delivery-status +Method: POST +Secret: (kirim terpisah lewat kanal aman — JANGAN taruh di email/chat biasa, + JANGAN commit ke dokumen/repo ini) +``` + +> ⚠️ URL di atas masih placeholder — isi dengan domain publik/staging service +> `omnix-broadcast` kami sebelum dikirim ke tim Omnix. Saat ini service jalan +> lokal di `localhost:8081`, belum bisa diakses dari luar. + +> 🔒 **Secret key** sudah kami generate (64 karakter hex, random 256-bit) dan +> tersimpan di `.env` service kami (`WEBHOOK_SECRET`). Kirim nilainya ke PIC +> Omnix lewat kanal aman (password manager, secret vault, atau chat terenkripsi +> — bukan email/Slack polos). Mereka pakai secret yang SAMA persis untuk +> menandatangani tiap request ke kami. + +## 3. Format payload — **dikonfirmasi tim Omnix** ✅ + +**Header:** +``` +X-Sopiga-Signature: +``` + +**Body:** +```json +{ + "recipient_detail_id": 1234, + "status": "sent", + "gateway": "628124878787" +} +``` + +| Field | Tipe | Keterangan | +|---|---|---| +| `recipient_detail_id` | integer | Sama dengan `recipient_detail_id` yang dikembalikan saat `add-recipient` | +| `status` | string | Salah satu: `pending`, `sent`, `delivered`, `read`, `failed` (atau `undelivered`) | +| `gateway` | string | Nomor WhatsApp pengirim (sender) | + +Sudah diuji end-to-end di sisi kami dengan payload persis seperti di atas — +`status: "sent"` diabaikan (belum final), `status: "delivered"`/`"read"` update +record jadi `delivered`, `status: "failed"`/`"undelivered"` update jadi `failed`. + +> Masih perlu dikonfirmasi: apakah header `X-Sopiga-Signature` (HMAC-SHA256) +> di atas juga dipakai Omnix, atau ada skema signature/auth lain di sisi mereka? + +## 4. Response yang kami kirim balik + +- `200 OK` — payload diterima & diproses. +- `400 Bad Request` — payload tidak valid/tidak bisa diparse. +- `401 Unauthorized` — signature tidak cocok. +- `500 Internal Server Error` — gagal proses di sisi kami (mohon di-retry). + +## 5. Fallback + +Selama callback belum aktif/terverifikasi, kami tetap jalankan polling manual +ke `GET /api/client/collar/add-recipient/{id}/detail` sebagai cadangan, jadi +tidak ada risiko data hilang selama masa transisi. diff --git a/domain/broadcast.go b/domain/broadcast.go new file mode 100644 index 0000000..5c44636 --- /dev/null +++ b/domain/broadcast.go @@ -0,0 +1,63 @@ +package domain + +import "time" + +type BroadcastStatus string + +const ( + StatusPending BroadcastStatus = "pending" + StatusDispatched BroadcastStatus = "dispatched" + StatusDelivered BroadcastStatus = "delivered" + StatusFailed BroadcastStatus = "failed" + StatusRetryScheduled BroadcastStatus = "retry_scheduled" +) + +type Broadcast struct { + ID int64 + SopigaCollarID int + SopigaTemplateID int + MessagePayload map[string]any + Status BroadcastStatus + ErrorMessage *string + ErrorCount int + SopigaRecipientDetailID *int64 + CreatedAt time.Time + UpdatedAt time.Time +} + +func (b *Broadcast) RecipientPhone() (string, bool) { + v, ok := b.MessagePayload["nasabah_phone"].(string) + return v, ok +} + +func (b *Broadcast) InvoiceURL() (string, bool) { + v, ok := b.MessagePayload["invoice_pdf_url"].(string) + return v, ok +} + +func (b *Broadcast) CustomerName() string { + v, _ := b.MessagePayload["nasabah_nama"].(string) + return v +} + +// MapSopigaDeliveryStatus translates a Sopiga recipient status into our +// internal terminal status. ok is false while Sopiga still reports an +// in-flight state (e.g. "sent"), meaning there is nothing to update yet. +func MapSopigaDeliveryStatus(sopigaStatus string) (status BroadcastStatus, ok bool) { + switch sopigaStatus { + case "delivered", "read": + return StatusDelivered, true + case "failed", "undelivered": + return StatusFailed, true + default: // pending, sent, ... + return "", false + } +} + +type TemplateVariable struct { + VariableOrder int + VariableName string + VariableType string + FieldSource string + IsRequired bool +} diff --git a/dto/sopiga.go b/dto/sopiga.go new file mode 100644 index 0000000..de13a46 --- /dev/null +++ b/dto/sopiga.go @@ -0,0 +1,54 @@ +package dto + +import "time" + +type CollarAddRecipientRequest struct { + BroadcastID int `json:"broadcast_id"` + TemplateID int `json:"template_id"` + Details CollarDetails `json:"details"` + Attachment *CollarAttachment `json:"attachment,omitempty"` + Labels map[string]string `json:"labels,omitempty"` +} + +type CollarDetails struct { + Recipient string `json:"recipient"` + Message string `json:"message"` +} + +type CollarAttachment struct { + Type string `json:"type"` + Caption string `json:"caption"` + File string `json:"file"` +} + +type CollarAPIResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + Data CollarResponseData `json:"data"` + Errors map[string][]string `json:"errors,omitempty"` +} + +type CollarResponseData struct { + BroadcastID int `json:"broadcast_id"` + RecipientDetailID int64 `json:"recipient_detail_id"` + Recipient string `json:"recipient"` + Status string `json:"status"` +} + +type CollarRecipientDetailResponse struct { + Success bool `json:"success"` + Message string `json:"message"` + Data CollarRecipientDetailData `json:"data"` +} + +type CollarRecipientDetailData struct { + ID int64 `json:"id"` + BroadcastID int `json:"broadcast_id"` + Recipient string `json:"recipient"` + TemplateID int `json:"template_id"` + Status string `json:"status"` // pending, sent, delivered, read, failed + ErrorMessage *string `json:"error_message"` + SentAt *time.Time `json:"sent_at"` + DeliveredAt *time.Time `json:"delivered_at"` + ReadAt *time.Time `json:"read_at"` +} diff --git a/dto/webhook.go b/dto/webhook.go new file mode 100644 index 0000000..3432436 --- /dev/null +++ b/dto/webhook.go @@ -0,0 +1,9 @@ +package dto + +// DeliveryStatusWebhook is the confirmed contract for an inbound Omnix +// callback when a recipient's delivery status changes. +type DeliveryStatusWebhook struct { + RecipientDetailID int64 `json:"recipient_detail_id"` + Status string `json:"status"` // pending, sent, delivered, read, failed + Gateway string `json:"gateway"` // sender WhatsApp number +} diff --git a/entity/broadcast.go b/entity/broadcast.go new file mode 100644 index 0000000..49a8e89 --- /dev/null +++ b/entity/broadcast.go @@ -0,0 +1,24 @@ +package entity + +import "time" + +type BroadcastStaging struct { + ID int64 `db:"id"` + SopigaCollarID int `db:"sopiga_collar_id"` + SopigaTemplateID int `db:"sopiga_template_id"` + MessagePayload []byte `db:"message_payload"` + Status string `db:"status"` + ErrorMessage *string `db:"error_message"` + ErrorCount int `db:"error_count"` + SopigaRecipientDetailID *int64 `db:"sopiga_recipient_detail_id"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} + +type TemplateVariableMapping struct { + VariableOrder int `db:"variable_order"` + SopigaVariableName string `db:"sopiga_variable_name"` + VariableType string `db:"variable_type"` + DBFieldSource string `db:"db_field_source"` + IsRequired bool `db:"is_required"` +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..71e7aae --- /dev/null +++ b/go.mod @@ -0,0 +1,20 @@ +module repository.promas.id/prana/omnix-sopiga + +go 1.25.0 + +require ( + github.com/jackc/pgx/v5 v5.5.0 + repository.promas.id/prana/go-dw-framework v0.0.0 +) + +require ( + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect + github.com/jackc/puddle/v2 v2.2.1 // indirect + github.com/stretchr/testify v1.8.3 // indirect + golang.org/x/crypto v0.48.0 // indirect + golang.org/x/sync v0.19.0 // indirect + golang.org/x/text v0.34.0 // indirect +) + +replace repository.promas.id/prana/go-dw-framework => ../go-dw-framework diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..2606adf --- /dev/null +++ b/go.sum @@ -0,0 +1,28 @@ +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a h1:bbPeKD0xmW/Y25WS6cokEszi5g+S0QxI/d45PkRi7Nk= +github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.5.0 h1:NxstgwndsTRy7eq9/kqYc/BZh5w2hHJV86wjvO+1xPw= +github.com/jackc/pgx/v5 v5.5.0/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA= +github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk= +github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.3 h1:RP3t2pwF7cMEbC1dqtB6poj3niw/9gnV4Cjg5oW5gtY= +github.com/stretchr/testify v1.8.3/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= +golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= +golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= +golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/handler/webhook.go b/handler/webhook.go new file mode 100644 index 0000000..89ed784 --- /dev/null +++ b/handler/webhook.go @@ -0,0 +1,82 @@ +package handler + +import ( + "crypto/hmac" + "crypto/sha256" + "crypto/subtle" + "encoding/hex" + "encoding/json" + "io" + "net/http" + + fwlogger "repository.promas.id/prana/go-dw-framework/logger" + + "repository.promas.id/prana/omnix-sopiga/dto" + "repository.promas.id/prana/omnix-sopiga/service" +) + +const signatureHeader = "X-Sopiga-Signature" + +// WebhookHandler receives delivery-status callbacks pushed by Omnix, +// as an alternative to polling. Payload shape confirmed by the Omnix team +// (see dto.DeliveryStatusWebhook); the signature scheme is still our own +// proposal pending their confirmation. +type WebhookHandler struct { + service *service.Service + logger *fwlogger.Logger + secret string +} + +func NewWebhookHandler(svc *service.Service, log *fwlogger.Logger, secret string) *WebhookHandler { + return &WebhookHandler{service: svc, logger: log, secret: secret} +} + +func (h *WebhookHandler) HandleDeliveryStatus(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + + body, err := io.ReadAll(r.Body) + if err != nil { + h.logger.Error("webhook: read body failed", "error", err) + w.WriteHeader(http.StatusBadRequest) + return + } + defer r.Body.Close() + + if h.secret != "" && !h.verifySignature(body, r.Header.Get(signatureHeader)) { + h.logger.Error("webhook: signature verification failed") + w.WriteHeader(http.StatusUnauthorized) + return + } + + var payload dto.DeliveryStatusWebhook + if err := json.Unmarshal(body, &payload); err != nil { + h.logger.Error("webhook: invalid payload", "error", err) + w.WriteHeader(http.StatusBadRequest) + return + } + + if err := h.service.SyncDeliveryFromWebhook(r.Context(), payload); err != nil { + h.logger.Error("webhook: sync delivery failed", + "recipient_detail_id", payload.RecipientDetailID, + "error", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + h.logger.Info("webhook: delivery status applied", + "recipient_detail_id", payload.RecipientDetailID, + "status", payload.Status, + "gateway", payload.Gateway) + + w.WriteHeader(http.StatusOK) +} + +func (h *WebhookHandler) verifySignature(body []byte, signature string) bool { + mac := hmac.New(sha256.New, []byte(h.secret)) + mac.Write(body) + expected := hex.EncodeToString(mac.Sum(nil)) + return subtle.ConstantTimeCompare([]byte(expected), []byte(signature)) == 1 +} diff --git a/main.go b/main.go new file mode 100644 index 0000000..fc393d5 --- /dev/null +++ b/main.go @@ -0,0 +1,83 @@ +package main + +import ( + "context" + "net/http" + "os" + "os/signal" + "syscall" + "time" + + fwconfig "repository.promas.id/prana/go-dw-framework/config" + fwdb "repository.promas.id/prana/go-dw-framework/db" + fwingestion "repository.promas.id/prana/go-dw-framework/ingestion" + fwlogger "repository.promas.id/prana/go-dw-framework/logger" + + "repository.promas.id/prana/omnix-sopiga/client" + "repository.promas.id/prana/omnix-sopiga/handler" + "repository.promas.id/prana/omnix-sopiga/repository" + "repository.promas.id/prana/omnix-sopiga/service" + "repository.promas.id/prana/omnix-sopiga/transformer" + "repository.promas.id/prana/omnix-sopiga/worker" +) + +func main() { + _ = fwconfig.LoadDotEnv(".env") + cfg := LoadConfig() + log := fwlogger.New("omnix-broadcast") + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + pool, err := fwdb.NewPool(ctx, fwdb.Config{ + DSN: cfg.DB.DSN, + MaxConns: cfg.DB.MaxConns, + MinConns: cfg.DB.MinConns, + MaxConnLifetime: cfg.DB.MaxConnLifetime, + MaxConnIdleTime: cfg.DB.MaxConnIdleTime, + }) + if err != nil { + log.Error("failed to connect to database", "error", err) + os.Exit(1) + } + defer pool.Close() + + repo := repository.New(pool) + tf := transformer.New() + sopigaClient := client.NewSopigaClient(cfg.Sopiga.BaseURL, cfg.Sopiga.Token) + retrier := fwingestion.NewRetrier(cfg.Worker.MaxRetries, cfg.Worker.RetryDelay) + + svc := service.New(repo, tf, sopigaClient, retrier, log) + bcWorker := worker.New(svc, log, cfg.Worker.CheckInterval, cfg.Worker.BatchSize) + syncWorker := worker.NewDeliverySync(svc, log, cfg.Worker.SyncCheckInterval, cfg.Worker.SyncBatchSize) + + go bcWorker.Start(ctx) + go syncWorker.Start(ctx) + + webhookHandler := handler.NewWebhookHandler(svc, log, cfg.Webhook.Secret) + mux := http.NewServeMux() + mux.HandleFunc("/webhooks/sopiga/delivery-status", webhookHandler.HandleDeliveryStatus) + + httpServer := &http.Server{ + Addr: ":" + cfg.Webhook.Port, + Handler: mux, + } + + go func() { + log.Info("webhook server listening", "port", cfg.Webhook.Port) + if err := httpServer.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Error("webhook server stopped", "error", err) + } + }() + + sigChan := make(chan os.Signal, 1) + signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM) + <-sigChan + + log.Info("shutting down gracefully") + cancel() + + shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer shutdownCancel() + _ = httpServer.Shutdown(shutdownCtx) +} diff --git a/migrations/001_create_collection_broadcasts.down.sql b/migrations/001_create_collection_broadcasts.down.sql new file mode 100644 index 0000000..9708a8b --- /dev/null +++ b/migrations/001_create_collection_broadcasts.down.sql @@ -0,0 +1,17 @@ +DROP FUNCTION IF EXISTS collection_broadcasts.update_broadcast_status(BIGINT, VARCHAR, TEXT, JSONB); +DROP FUNCTION IF EXISTS collection_broadcasts.get_template_variables(INT); + +DROP VIEW IF EXISTS collection_broadcasts.v_delivery_rate_24h; +DROP VIEW IF EXISTS collection_broadcasts.v_failed_records_24h; +DROP VIEW IF EXISTS collection_broadcasts.v_collar_summary; +DROP VIEW IF EXISTS collection_broadcasts.v_template_variables_ordered; + +DROP TABLE IF EXISTS collection_broadcasts.sopiga_sync_job; +DROP TABLE IF EXISTS collection_broadcasts.broadcast_error_log; +DROP TABLE IF EXISTS collection_broadcasts.broadcast_audit_log; +DROP TABLE IF EXISTS collection_broadcasts.broadcast_staging; +DROP TABLE IF EXISTS collection_broadcasts.sopiga_collar_config; +DROP TABLE IF EXISTS collection_broadcasts.template_variable_mapping; +DROP TABLE IF EXISTS collection_broadcasts.sopiga_template_config; + +DROP SCHEMA IF EXISTS collection_broadcasts; diff --git a/migrations/001_create_collection_broadcasts.up.sql b/migrations/001_create_collection_broadcasts.up.sql new file mode 100644 index 0000000..3b08b3c --- /dev/null +++ b/migrations/001_create_collection_broadcasts.up.sql @@ -0,0 +1,258 @@ +CREATE SCHEMA IF NOT EXISTS collection_broadcasts; + +-- ============================================================================ +-- 1. CONFIGURATION TABLES (Reference only) +-- ============================================================================ + +CREATE TABLE collection_broadcasts.sopiga_template_config ( + id BIGSERIAL PRIMARY KEY, + template_name VARCHAR(100) NOT NULL, + sopiga_template_id INT NOT NULL UNIQUE, + template_type VARCHAR(50), + channel VARCHAR(50), + description TEXT, + active BOOLEAN DEFAULT TRUE, + created_at TIMESTAMP DEFAULT NOW(), + updated_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_sopiga_template_active ON collection_broadcasts.sopiga_template_config(active); + +CREATE TABLE collection_broadcasts.template_variable_mapping ( + id BIGSERIAL PRIMARY KEY, + sopiga_template_id INT NOT NULL REFERENCES collection_broadcasts.sopiga_template_config(sopiga_template_id) ON DELETE CASCADE, + variable_order INT NOT NULL, + sopiga_variable_name VARCHAR(100) NOT NULL, + variable_type VARCHAR(50), + db_field_source VARCHAR(100) NOT NULL, + is_required BOOLEAN DEFAULT TRUE, + example_value VARCHAR(500), + description TEXT, + created_at TIMESTAMP DEFAULT NOW(), + updated_at TIMESTAMP DEFAULT NOW(), + + CONSTRAINT unique_template_variable UNIQUE (sopiga_template_id, sopiga_variable_name), + CONSTRAINT unique_variable_order UNIQUE (sopiga_template_id, variable_order) +); + +CREATE INDEX idx_template_variable_mapping_template_id ON collection_broadcasts.template_variable_mapping(sopiga_template_id); +CREATE INDEX idx_template_variable_mapping_order ON collection_broadcasts.template_variable_mapping(sopiga_template_id, variable_order); + +CREATE TABLE collection_broadcasts.sopiga_collar_config ( + id BIGSERIAL PRIMARY KEY, + broadcast_name VARCHAR(100) NOT NULL, + sopiga_collar_id INT NOT NULL UNIQUE, + sopiga_template_id INT NOT NULL REFERENCES collection_broadcasts.sopiga_template_config(sopiga_template_id), + status VARCHAR(50), + description TEXT, + created_at TIMESTAMP DEFAULT NOW(), + updated_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_sopiga_collar_status ON collection_broadcasts.sopiga_collar_config(status); + +-- ============================================================================ +-- 2. BROADCAST STAGING (CORE) - DENORMALIZED +-- ============================================================================ + +CREATE TABLE collection_broadcasts.broadcast_staging ( + id BIGSERIAL PRIMARY KEY, + + sopiga_collar_id INT NOT NULL REFERENCES collection_broadcasts.sopiga_collar_config(sopiga_collar_id), + sopiga_template_id INT NOT NULL REFERENCES collection_broadcasts.sopiga_template_config(sopiga_template_id), + + -- Single source of truth: all fields needed to build the message live here + message_payload JSONB NOT NULL, + + status VARCHAR(50) DEFAULT 'pending' NOT NULL, + error_message TEXT, + error_count INT DEFAULT 0, + + sopiga_recipient_detail_id BIGINT, + sopiga_external_id VARCHAR(100), + + created_at TIMESTAMP DEFAULT NOW(), + updated_at TIMESTAMP DEFAULT NOW(), + dispatched_at TIMESTAMP, + delivered_at TIMESTAMP, + failed_at TIMESTAMP, + + CONSTRAINT status_valid CHECK (status IN ('pending', 'dispatched', 'delivered', 'failed', 'retry_scheduled')), + CONSTRAINT error_count_positive CHECK (error_count >= 0) +); + +CREATE INDEX idx_broadcast_staging_status ON collection_broadcasts.broadcast_staging(status); +CREATE INDEX idx_broadcast_staging_created_at ON collection_broadcasts.broadcast_staging(created_at DESC); +CREATE INDEX idx_broadcast_staging_collar_id ON collection_broadcasts.broadcast_staging(sopiga_collar_id); +CREATE INDEX idx_broadcast_staging_template_id ON collection_broadcasts.broadcast_staging(sopiga_template_id); + +CREATE INDEX idx_broadcast_staging_pending_query + ON collection_broadcasts.broadcast_staging(status, created_at ASC) + WHERE status = 'pending'; + +CREATE INDEX idx_broadcast_staging_retry_query + ON collection_broadcasts.broadcast_staging(status, error_count, created_at ASC) + WHERE status = 'retry_scheduled' AND error_count < 3; + +-- ============================================================================ +-- 3. AUDIT & LOGGING +-- ============================================================================ + +CREATE TABLE collection_broadcasts.broadcast_audit_log ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + old_status VARCHAR(50), + new_status VARCHAR(50) NOT NULL, + reason VARCHAR(500), + sopiga_response JSONB, + changed_by VARCHAR(100) DEFAULT 'system', + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_broadcast_audit_log_broadcast_id ON collection_broadcasts.broadcast_audit_log(broadcast_id); +CREATE INDEX idx_broadcast_audit_log_created_at ON collection_broadcasts.broadcast_audit_log(created_at DESC); + +CREATE TABLE collection_broadcasts.broadcast_error_log ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + error_type VARCHAR(100), + error_code VARCHAR(50), + error_message TEXT, + error_details JSONB, + attempt_number INT, + next_retry_at TIMESTAMP, + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_broadcast_error_log_broadcast_id ON collection_broadcasts.broadcast_error_log(broadcast_id); +CREATE INDEX idx_broadcast_error_log_error_type ON collection_broadcasts.broadcast_error_log(error_type); + +CREATE TABLE collection_broadcasts.sopiga_sync_job ( + id BIGSERIAL PRIMARY KEY, + broadcast_id BIGINT NOT NULL REFERENCES collection_broadcasts.broadcast_staging(id) ON DELETE CASCADE, + sopiga_recipient_detail_id BIGINT, + last_synced_at TIMESTAMP, + last_status_from_sopiga VARCHAR(50), + sync_count INT DEFAULT 0, + next_sync_at TIMESTAMP, + completed_at TIMESTAMP, + created_at TIMESTAMP DEFAULT NOW() +); + +CREATE INDEX idx_sopiga_sync_job_broadcast_id ON collection_broadcasts.sopiga_sync_job(broadcast_id); +CREATE INDEX idx_sopiga_sync_job_next_sync_at ON collection_broadcasts.sopiga_sync_job(next_sync_at); + +-- ============================================================================ +-- 4. VIEWS +-- ============================================================================ + +CREATE VIEW collection_broadcasts.v_template_variables_ordered AS +SELECT + tvm.sopiga_template_id, + stc.template_name, + tvm.variable_order, + tvm.sopiga_variable_name, + tvm.variable_type, + tvm.db_field_source, + tvm.is_required, + tvm.example_value +FROM collection_broadcasts.template_variable_mapping tvm +JOIN collection_broadcasts.sopiga_template_config stc ON tvm.sopiga_template_id = stc.sopiga_template_id +WHERE stc.active = TRUE +ORDER BY tvm.sopiga_template_id, tvm.variable_order; + +CREATE VIEW collection_broadcasts.v_collar_summary AS +SELECT + sc.sopiga_collar_id, + sc.broadcast_name, + COUNT(*) as total_records, + COUNT(CASE WHEN bs.status = 'pending' THEN 1 END) as pending, + COUNT(CASE WHEN bs.status = 'dispatched' THEN 1 END) as dispatched, + COUNT(CASE WHEN bs.status = 'delivered' THEN 1 END) as delivered, + COUNT(CASE WHEN bs.status = 'failed' THEN 1 END) as failed, + ROUND(100.0 * COUNT(CASE WHEN bs.status = 'delivered' THEN 1 END) / NULLIF(COUNT(*), 0), 2) as delivery_rate_percent +FROM collection_broadcasts.sopiga_collar_config sc +LEFT JOIN collection_broadcasts.broadcast_staging bs ON sc.sopiga_collar_id = bs.sopiga_collar_id +GROUP BY sc.sopiga_collar_id, sc.broadcast_name; + +CREATE VIEW collection_broadcasts.v_failed_records_24h AS +SELECT + id, + sopiga_collar_id, + sopiga_template_id, + message_payload->>'nasabah_nama' as nasabah_nama, + message_payload->>'nasabah_phone' as nasabah_phone, + error_message, + error_count, + failed_at, + created_at +FROM collection_broadcasts.broadcast_staging +WHERE status = 'failed' AND created_at > NOW() - INTERVAL '1 DAY' +ORDER BY failed_at DESC; + +CREATE VIEW collection_broadcasts.v_delivery_rate_24h AS +SELECT + COUNT(*) as total, + COUNT(CASE WHEN status = 'delivered' THEN 1 END) as delivered, + COUNT(CASE WHEN status = 'failed' THEN 1 END) as failed, + COUNT(CASE WHEN status = 'dispatched' THEN 1 END) as in_progress, + ROUND(100.0 * COUNT(CASE WHEN status = 'delivered' THEN 1 END) / NULLIF(COUNT(*), 0), 2) as delivery_rate_percent +FROM collection_broadcasts.broadcast_staging +WHERE created_at > NOW() - INTERVAL '1 DAY'; + +-- ============================================================================ +-- 5. HELPER FUNCTIONS +-- ============================================================================ + +CREATE OR REPLACE FUNCTION collection_broadcasts.get_template_variables( + p_sopiga_template_id INT +) +RETURNS TABLE( + variable_order INT, + sopiga_variable_name VARCHAR, + variable_type VARCHAR, + db_field_source VARCHAR, + is_required BOOLEAN +) AS $$ +BEGIN + RETURN QUERY + SELECT + tvm.variable_order, + tvm.sopiga_variable_name, + tvm.variable_type, + tvm.db_field_source, + tvm.is_required + FROM collection_broadcasts.template_variable_mapping tvm + WHERE tvm.sopiga_template_id = p_sopiga_template_id + ORDER BY tvm.variable_order ASC; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION collection_broadcasts.update_broadcast_status( + p_broadcast_id BIGINT, + p_new_status VARCHAR, + p_error_message TEXT DEFAULT NULL, + p_sopiga_response JSONB DEFAULT NULL +) +RETURNS VOID AS $$ +DECLARE + v_old_status VARCHAR; +BEGIN + SELECT status INTO v_old_status + FROM collection_broadcasts.broadcast_staging + WHERE id = p_broadcast_id; + + UPDATE collection_broadcasts.broadcast_staging + SET + status = p_new_status, + error_message = p_error_message, + updated_at = NOW(), + dispatched_at = CASE WHEN p_new_status = 'dispatched' THEN NOW() ELSE dispatched_at END, + delivered_at = CASE WHEN p_new_status = 'delivered' THEN NOW() ELSE delivered_at END, + failed_at = CASE WHEN p_new_status = 'failed' THEN NOW() ELSE failed_at END + WHERE id = p_broadcast_id; + + INSERT INTO collection_broadcasts.broadcast_audit_log (broadcast_id, old_status, new_status, sopiga_response) + VALUES (p_broadcast_id, v_old_status, p_new_status, p_sopiga_response); +END; +$$ LANGUAGE plpgsql; diff --git a/repository/broadcast.go b/repository/broadcast.go new file mode 100644 index 0000000..09bf1b5 --- /dev/null +++ b/repository/broadcast.go @@ -0,0 +1,171 @@ +package repository + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/jackc/pgx/v5/pgxpool" + + "repository.promas.id/prana/omnix-sopiga/entity" +) + +type Repository struct { + pool *pgxpool.Pool +} + +func New(pool *pgxpool.Pool) *Repository { + return &Repository{pool: pool} +} + +func (r *Repository) FindPending(ctx context.Context, limit int) ([]*entity.BroadcastStaging, error) { + query := ` + SELECT + id, sopiga_collar_id, sopiga_template_id, message_payload, + status, error_message, error_count, sopiga_recipient_detail_id, + created_at, updated_at + FROM collection_broadcasts.broadcast_staging + WHERE status = 'pending' + ORDER BY created_at ASC + LIMIT $1 + ` + + rows, err := r.pool.Query(ctx, query, limit) + if err != nil { + return nil, fmt.Errorf("find pending: %w", err) + } + defer rows.Close() + + var records []*entity.BroadcastStaging + for rows.Next() { + var e entity.BroadcastStaging + if err := rows.Scan( + &e.ID, &e.SopigaCollarID, &e.SopigaTemplateID, &e.MessagePayload, + &e.Status, &e.ErrorMessage, &e.ErrorCount, &e.SopigaRecipientDetailID, + &e.CreatedAt, &e.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + records = append(records, &e) + } + + return records, rows.Err() +} + +func (r *Repository) FindTemplateVariables(ctx context.Context, templateID int) ([]*entity.TemplateVariableMapping, error) { + query := ` + SELECT variable_order, sopiga_variable_name, variable_type, db_field_source, is_required + FROM collection_broadcasts.template_variable_mapping + WHERE sopiga_template_id = $1 + ORDER BY variable_order ASC + ` + + rows, err := r.pool.Query(ctx, query, templateID) + if err != nil { + return nil, fmt.Errorf("find template variables: %w", err) + } + defer rows.Close() + + var variables []*entity.TemplateVariableMapping + for rows.Next() { + var v entity.TemplateVariableMapping + if err := rows.Scan(&v.VariableOrder, &v.SopigaVariableName, &v.VariableType, &v.DBFieldSource, &v.IsRequired); err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + variables = append(variables, &v) + } + + return variables, rows.Err() +} + +func (r *Repository) FindByRecipientDetailID(ctx context.Context, recipientDetailID int64) (*entity.BroadcastStaging, error) { + query := ` + SELECT + id, sopiga_collar_id, sopiga_template_id, message_payload, + status, error_message, error_count, sopiga_recipient_detail_id, + created_at, updated_at + FROM collection_broadcasts.broadcast_staging + WHERE sopiga_recipient_detail_id = $1 + ` + + var e entity.BroadcastStaging + err := r.pool.QueryRow(ctx, query, recipientDetailID).Scan( + &e.ID, &e.SopigaCollarID, &e.SopigaTemplateID, &e.MessagePayload, + &e.Status, &e.ErrorMessage, &e.ErrorCount, &e.SopigaRecipientDetailID, + &e.CreatedAt, &e.UpdatedAt, + ) + if err != nil { + return nil, fmt.Errorf("find by recipient_detail_id %d: %w", recipientDetailID, err) + } + + return &e, nil +} + +func (r *Repository) UpdateStatus(ctx context.Context, id int64, status, errMsg string, sopigaResponse any) error { + var respJSON []byte + if sopigaResponse != nil { + var err error + respJSON, err = json.Marshal(sopigaResponse) + if err != nil { + return fmt.Errorf("marshal sopiga response: %w", err) + } + } + + query := `SELECT collection_broadcasts.update_broadcast_status($1, $2, $3, $4)` + _, err := r.pool.Exec(ctx, query, id, status, errMsg, respJSON) + if err != nil { + return fmt.Errorf("update status: %w", err) + } + return nil +} + +func (r *Repository) FindDispatched(ctx context.Context, limit int) ([]*entity.BroadcastStaging, error) { + query := ` + SELECT + id, sopiga_collar_id, sopiga_template_id, message_payload, + status, error_message, error_count, sopiga_recipient_detail_id, + created_at, updated_at + FROM collection_broadcasts.broadcast_staging + WHERE status = 'dispatched' AND sopiga_recipient_detail_id IS NOT NULL + ORDER BY dispatched_at ASC + LIMIT $1 + ` + + rows, err := r.pool.Query(ctx, query, limit) + if err != nil { + return nil, fmt.Errorf("find dispatched: %w", err) + } + defer rows.Close() + + var records []*entity.BroadcastStaging + for rows.Next() { + var e entity.BroadcastStaging + if err := rows.Scan( + &e.ID, &e.SopigaCollarID, &e.SopigaTemplateID, &e.MessagePayload, + &e.Status, &e.ErrorMessage, &e.ErrorCount, &e.SopigaRecipientDetailID, + &e.CreatedAt, &e.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("scan: %w", err) + } + records = append(records, &e) + } + + return records, rows.Err() +} + +func (r *Repository) UpdateDispatched(ctx context.Context, id int64, recipientDetailID int64) error { + query := ` + UPDATE collection_broadcasts.broadcast_staging + SET + status = 'dispatched', + sopiga_recipient_detail_id = $1, + dispatched_at = NOW(), + updated_at = NOW() + WHERE id = $2 + ` + _, err := r.pool.Exec(ctx, query, recipientDetailID, id) + if err != nil { + return fmt.Errorf("update dispatched: %w", err) + } + return nil +} diff --git a/service/broadcast.go b/service/broadcast.go new file mode 100644 index 0000000..bac27f3 --- /dev/null +++ b/service/broadcast.go @@ -0,0 +1,205 @@ +package service + +import ( + "context" + "fmt" + + fwingestion "repository.promas.id/prana/go-dw-framework/ingestion" + fwlogger "repository.promas.id/prana/go-dw-framework/logger" + + "repository.promas.id/prana/omnix-sopiga/client" + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/dto" + "repository.promas.id/prana/omnix-sopiga/repository" + "repository.promas.id/prana/omnix-sopiga/transformer" + "repository.promas.id/prana/omnix-sopiga/validator" +) + +type Service struct { + repository *repository.Repository + transformer *transformer.Transformer + sopiga *client.SopigaClient + retrier *fwingestion.Retrier + logger *fwlogger.Logger +} + +func New( + repo *repository.Repository, + tf *transformer.Transformer, + sopigaClient *client.SopigaClient, + retrier *fwingestion.Retrier, + log *fwlogger.Logger, +) *Service { + return &Service{ + repository: repo, + transformer: tf, + sopiga: sopigaClient, + retrier: retrier, + logger: log, + } +} + +// FetchPending returns the next batch of broadcasts waiting to be dispatched. +func (s *Service) FetchPending(ctx context.Context, limit int) ([]*domain.Broadcast, error) { + entities, err := s.repository.FindPending(ctx, limit) + if err != nil { + return nil, fmt.Errorf("fetch pending: %w", err) + } + + broadcasts := make([]*domain.Broadcast, 0, len(entities)) + for _, e := range entities { + b, err := s.transformer.EntityToDomain(e) + if err != nil { + s.logger.Error("skip malformed record", "broadcast_id", e.ID, "error", err) + continue + } + broadcasts = append(broadcasts, b) + } + + return broadcasts, nil +} + +// Dispatch builds the templated message for a broadcast and sends it to Sopiga, +// updating status in the database based on the outcome. +func (s *Service) Dispatch(ctx context.Context, b *domain.Broadcast) error { + if err := validator.ValidateBroadcastPayload(b); err != nil { + s.markFailed(ctx, b.ID, err) + return err + } + + variableEntities, err := s.repository.FindTemplateVariables(ctx, b.SopigaTemplateID) + if err != nil { + s.markFailed(ctx, b.ID, err) + return fmt.Errorf("dispatch %d: %w", b.ID, err) + } + + variables := make([]*domain.TemplateVariable, len(variableEntities)) + for i, ve := range variableEntities { + variables[i] = s.transformer.EntityToDomainVariable(ve) + } + + message, err := s.transformer.BuildDynamicMessage(variables, b.MessagePayload) + if err != nil { + s.markFailed(ctx, b.ID, err) + return fmt.Errorf("dispatch %d: %w", b.ID, err) + } + + recipientPhone, _ := b.RecipientPhone() + invoiceURL, _ := b.InvoiceURL() + + req := dto.CollarAddRecipientRequest{ + BroadcastID: b.SopigaCollarID, + TemplateID: b.SopigaTemplateID, + Details: dto.CollarDetails{ + Recipient: recipientPhone, + Message: message, + }, + Attachment: &dto.CollarAttachment{ + Type: "document", + Caption: fmt.Sprintf("Invoice - %s", b.CustomerName()), + File: invoiceURL, + }, + Labels: map[string]string{"module": "gadai_collection"}, + } + + resp, err := s.retrier.Do(ctx, func() (int64, error) { + r, err := s.sopiga.AddRecipient(ctx, req) + if err != nil { + return 0, err + } + if !r.Success { + return 0, fmt.Errorf("sopiga error: %s", r.Message) + } + return r.Data.RecipientDetailID, nil + }) + if err != nil { + s.markFailed(ctx, b.ID, err) + return fmt.Errorf("dispatch %d: %w", b.ID, err) + } + + if err := s.repository.UpdateDispatched(ctx, b.ID, resp); err != nil { + return fmt.Errorf("dispatch %d: %w", b.ID, err) + } + + return nil +} + +// FetchDispatched returns broadcasts awaiting a delivery status update from Sopiga. +func (s *Service) FetchDispatched(ctx context.Context, limit int) ([]*domain.Broadcast, error) { + entities, err := s.repository.FindDispatched(ctx, limit) + if err != nil { + return nil, fmt.Errorf("fetch dispatched: %w", err) + } + + broadcasts := make([]*domain.Broadcast, 0, len(entities)) + for _, e := range entities { + b, err := s.transformer.EntityToDomain(e) + if err != nil { + s.logger.Error("skip malformed record", "broadcast_id", e.ID, "error", err) + continue + } + broadcasts = append(broadcasts, b) + } + + return broadcasts, nil +} + +// SyncDelivery checks the current delivery status of a dispatched broadcast +// with Sopiga and, if it has reached a terminal state, updates our record. +func (s *Service) SyncDelivery(ctx context.Context, b *domain.Broadcast) error { + if b.SopigaRecipientDetailID == nil { + return fmt.Errorf("sync delivery %d: missing recipient_detail_id", b.ID) + } + + resp, err := s.sopiga.GetRecipientDetail(ctx, *b.SopigaRecipientDetailID) + if err != nil { + return fmt.Errorf("sync delivery %d: %w", b.ID, err) + } + + status, ok := domain.MapSopigaDeliveryStatus(resp.Data.Status) + if !ok { + return nil // still in-flight (e.g. "sent"), nothing to update yet + } + + errMsg := "" + if resp.Data.ErrorMessage != nil { + errMsg = *resp.Data.ErrorMessage + } + + if err := s.repository.UpdateStatus(ctx, b.ID, string(status), errMsg, resp.Data); err != nil { + return fmt.Errorf("sync delivery %d: %w", b.ID, err) + } + + return nil +} + +// SyncDeliveryFromWebhook applies a delivery status pushed by an Omnix +// callback, avoiding the extra GET round-trip the polling sync worker needs. +func (s *Service) SyncDeliveryFromWebhook(ctx context.Context, payload dto.DeliveryStatusWebhook) error { + e, err := s.repository.FindByRecipientDetailID(ctx, payload.RecipientDetailID) + if err != nil { + return fmt.Errorf("sync delivery from webhook: %w", err) + } + + status, ok := domain.MapSopigaDeliveryStatus(payload.Status) + if !ok { + return nil // in-flight status (e.g. "sent"), nothing to update yet + } + + errMsg := "" + if status == domain.StatusFailed { + errMsg = fmt.Sprintf("delivery failed via gateway %s", payload.Gateway) + } + + if err := s.repository.UpdateStatus(ctx, e.ID, string(status), errMsg, payload); err != nil { + return fmt.Errorf("sync delivery from webhook %d: %w", e.ID, err) + } + + return nil +} + +func (s *Service) markFailed(ctx context.Context, id int64, cause error) { + if err := s.repository.UpdateStatus(ctx, id, string(domain.StatusFailed), cause.Error(), nil); err != nil { + s.logger.Error("failed to mark broadcast failed", "broadcast_id", id, "error", err) + } +} diff --git a/tests/transformer_test.go b/tests/transformer_test.go new file mode 100644 index 0000000..a4af72c --- /dev/null +++ b/tests/transformer_test.go @@ -0,0 +1,47 @@ +package tests + +import ( + "testing" + + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/transformer" +) + +func TestBuildDynamicMessage_Ordered(t *testing.T) { + tf := transformer.New() + + variables := []*domain.TemplateVariable{ + {VariableOrder: 1, VariableName: "Nama", VariableType: "string", FieldSource: "nasabah_nama", IsRequired: true}, + {VariableOrder: 2, VariableName: "TotalTagihan", VariableType: "integer", FieldSource: "nominal_tagihan", IsRequired: true}, + {VariableOrder: 3, VariableName: "TanggalJatuhTempo", VariableType: "date", FieldSource: "tanggal_tempo", IsRequired: true}, + } + + payload := map[string]any{ + "nasabah_nama": "Budi Santoso", + "nominal_tagihan": 1500000.0, + "tanggal_tempo": "2026-06-30", + } + + message, err := tf.BuildDynamicMessage(variables, payload) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + + want := "Budi Santoso#1500000#2026-06-30" + if message != want { + t.Fatalf("expected %q, got %q", want, message) + } +} + +func TestBuildDynamicMessage_MissingRequiredField(t *testing.T) { + tf := transformer.New() + + variables := []*domain.TemplateVariable{ + {VariableOrder: 1, VariableName: "Nama", VariableType: "string", FieldSource: "nasabah_nama", IsRequired: true}, + } + + _, err := tf.BuildDynamicMessage(variables, map[string]any{}) + if err == nil { + t.Fatal("expected error for missing required field, got nil") + } +} diff --git a/tests/validator_test.go b/tests/validator_test.go new file mode 100644 index 0000000..1d93ced --- /dev/null +++ b/tests/validator_test.go @@ -0,0 +1,33 @@ +package tests + +import ( + "testing" + + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/validator" +) + +func TestValidateBroadcastPayload_MissingPhone(t *testing.T) { + b := &domain.Broadcast{ + MessagePayload: map[string]any{ + "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf", + }, + } + + if err := validator.ValidateBroadcastPayload(b); err != validator.ErrMissingRecipientPhone { + t.Fatalf("expected ErrMissingRecipientPhone, got %v", err) + } +} + +func TestValidateBroadcastPayload_Valid(t *testing.T) { + b := &domain.Broadcast{ + MessagePayload: map[string]any{ + "nasabah_phone": "6281234567890", + "invoice_pdf_url": "https://storage.gadai.com/invoices/inv-001.pdf", + }, + } + + if err := validator.ValidateBroadcastPayload(b); err != nil { + t.Fatalf("expected no error, got %v", err) + } +} diff --git a/transformer/broadcast.go b/transformer/broadcast.go new file mode 100644 index 0000000..7cbca44 --- /dev/null +++ b/transformer/broadcast.go @@ -0,0 +1,76 @@ +package transformer + +import ( + "encoding/json" + "fmt" + "strings" + + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/entity" +) + +type Transformer struct{} + +func New() *Transformer { + return &Transformer{} +} + +func (t *Transformer) EntityToDomain(e *entity.BroadcastStaging) (*domain.Broadcast, error) { + var payload map[string]any + if err := json.Unmarshal(e.MessagePayload, &payload); err != nil { + return nil, fmt.Errorf("unmarshal message_payload: %w", err) + } + + return &domain.Broadcast{ + ID: e.ID, + SopigaCollarID: e.SopigaCollarID, + SopigaTemplateID: e.SopigaTemplateID, + MessagePayload: payload, + Status: domain.BroadcastStatus(e.Status), + ErrorMessage: e.ErrorMessage, + ErrorCount: e.ErrorCount, + SopigaRecipientDetailID: e.SopigaRecipientDetailID, + CreatedAt: e.CreatedAt, + UpdatedAt: e.UpdatedAt, + }, nil +} + +func (t *Transformer) EntityToDomainVariable(e *entity.TemplateVariableMapping) *domain.TemplateVariable { + return &domain.TemplateVariable{ + VariableOrder: e.VariableOrder, + VariableName: e.SopigaVariableName, + VariableType: e.VariableType, + FieldSource: e.DBFieldSource, + IsRequired: e.IsRequired, + } +} + +// BuildDynamicMessage interpolates the message payload against the ordered +// template variables, joining formatted values with "#" — the delimiter +// Sopiga expects between positional template placeholders. +func (t *Transformer) BuildDynamicMessage(variables []*domain.TemplateVariable, payload map[string]any) (string, error) { + parts := make([]string, 0, len(variables)) + + for _, v := range variables { + value, exists := payload[v.FieldSource] + if !exists { + if v.IsRequired { + return "", fmt.Errorf("missing required field: %s", v.FieldSource) + } + value = "" + } + + parts = append(parts, formatValue(v.VariableType, value)) + } + + return strings.Join(parts, "#"), nil +} + +func formatValue(variableType string, value any) string { + switch variableType { + case "integer", "decimal": + return fmt.Sprintf("%.0f", value) + default: + return fmt.Sprintf("%v", value) + } +} diff --git a/validator/broadcast.go b/validator/broadcast.go new file mode 100644 index 0000000..fb90169 --- /dev/null +++ b/validator/broadcast.go @@ -0,0 +1,22 @@ +package validator + +import ( + "errors" + + "repository.promas.id/prana/omnix-sopiga/domain" +) + +var ( + ErrMissingRecipientPhone = errors.New("nasabah_phone not in payload") + ErrMissingInvoiceURL = errors.New("invoice_pdf_url not in payload") +) + +func ValidateBroadcastPayload(b *domain.Broadcast) error { + if _, ok := b.RecipientPhone(); !ok { + return ErrMissingRecipientPhone + } + if _, ok := b.InvoiceURL(); !ok { + return ErrMissingInvoiceURL + } + return nil +} diff --git a/worker/broadcast_worker.go b/worker/broadcast_worker.go new file mode 100644 index 0000000..0a30690 --- /dev/null +++ b/worker/broadcast_worker.go @@ -0,0 +1,85 @@ +package worker + +import ( + "context" + "time" + + fwlogger "repository.promas.id/prana/go-dw-framework/logger" + + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/service" +) + +const maxConcurrency = 5 + +type BroadcastWorker struct { + service *service.Service + logger *fwlogger.Logger + checkInterval time.Duration + batchSize int +} + +func New(svc *service.Service, log *fwlogger.Logger, checkInterval time.Duration, batchSize int) *BroadcastWorker { + return &BroadcastWorker{ + service: svc, + logger: log, + checkInterval: checkInterval, + batchSize: batchSize, + } +} + +func (w *BroadcastWorker) Start(ctx context.Context) { + w.logger.Info("broadcast worker started") + + ticker := time.NewTicker(w.checkInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + w.logger.Info("broadcast worker stopped") + return + case <-ticker.C: + w.processPendingRecords(ctx) + } + } +} + +func (w *BroadcastWorker) processPendingRecords(ctx context.Context) { + broadcasts, err := w.service.FetchPending(ctx, w.batchSize) + if err != nil { + w.logger.Error("fetch pending failed", "error", err) + return + } + if len(broadcasts) == 0 { + return + } + + w.logger.Info("processing pending broadcasts", "count", len(broadcasts)) + + sem := make(chan struct{}, maxConcurrency) + done := make(chan struct{}, len(broadcasts)) + + for _, b := range broadcasts { + sem <- struct{}{} + go func(b *domain.Broadcast) { + defer func() { + <-sem + done <- struct{}{} + }() + w.dispatchOne(ctx, b) + }(b) + } + + for i := 0; i < len(broadcasts); i++ { + <-done + } +} + +func (w *BroadcastWorker) dispatchOne(ctx context.Context, b *domain.Broadcast) { + if err := w.service.Dispatch(ctx, b); err != nil { + w.logger.Error("dispatch failed", "broadcast_id", b.ID, "error", err) + return + } + w.logger.Info("broadcast dispatched", "broadcast_id", b.ID) +} diff --git a/worker/delivery_sync_worker.go b/worker/delivery_sync_worker.go new file mode 100644 index 0000000..4cbdae7 --- /dev/null +++ b/worker/delivery_sync_worker.go @@ -0,0 +1,83 @@ +package worker + +import ( + "context" + "time" + + fwlogger "repository.promas.id/prana/go-dw-framework/logger" + + "repository.promas.id/prana/omnix-sopiga/domain" + "repository.promas.id/prana/omnix-sopiga/service" +) + +type DeliverySyncWorker struct { + service *service.Service + logger *fwlogger.Logger + checkInterval time.Duration + batchSize int +} + +func NewDeliverySync(svc *service.Service, log *fwlogger.Logger, checkInterval time.Duration, batchSize int) *DeliverySyncWorker { + return &DeliverySyncWorker{ + service: svc, + logger: log, + checkInterval: checkInterval, + batchSize: batchSize, + } +} + +func (w *DeliverySyncWorker) Start(ctx context.Context) { + w.logger.Info("delivery sync worker started") + + ticker := time.NewTicker(w.checkInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + w.logger.Info("delivery sync worker stopped") + return + case <-ticker.C: + w.syncDispatchedRecords(ctx) + } + } +} + +func (w *DeliverySyncWorker) syncDispatchedRecords(ctx context.Context) { + broadcasts, err := w.service.FetchDispatched(ctx, w.batchSize) + if err != nil { + w.logger.Error("fetch dispatched failed", "error", err) + return + } + if len(broadcasts) == 0 { + return + } + + w.logger.Info("syncing delivery status", "count", len(broadcasts)) + + sem := make(chan struct{}, maxConcurrency) + done := make(chan struct{}, len(broadcasts)) + + for _, b := range broadcasts { + sem <- struct{}{} + go func(b *domain.Broadcast) { + defer func() { + <-sem + done <- struct{}{} + }() + w.syncOne(ctx, b) + }(b) + } + + for i := 0; i < len(broadcasts); i++ { + <-done + } +} + +func (w *DeliverySyncWorker) syncOne(ctx context.Context, b *domain.Broadcast) { + if err := w.service.SyncDelivery(ctx, b); err != nil { + w.logger.Error("sync delivery failed", "broadcast_id", b.ID, "error", err) + return + } + w.logger.Info("delivery status synced", "broadcast_id", b.ID) +}