prana/omnix-sopiga
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: extract omnix-broadcast as standalone repo omnix-sopiga

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
pranajaya161 2026-08-07 12:36:57 +07:00
commit 21780412b8
30 changed files with 3402 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
.env.example Normal file
View File

@ -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=

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
.go-cache/
*.log
.env
.DS_Store
# compiled binary from `go build`
/omnix-sopiga
/omnix-sopiga.exe

51
.gitlab-ci.yml Normal file
View File

@ -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"'

36
Dockerfile Normal file
View File

@ -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"]

124
README.md Normal file
View File

@ -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-remote> ../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
```

107
client/sopiga.go Normal file
View File

@ -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
}

69
config.go Normal file
View File

@ -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", ""),
},
}
}

452
docs/Collection_broadcast_worker_simplified.go Normal file
View File

@ -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<<uint(attempt)) * time.Second
w.logger.Printf("Retry attempt %d/%d in %v", attempt+1, w.maxRetries, waitTime)
time.Sleep(waitTime)
return w.callCollarAPIWithRetry(ctx, req, attempt+1)
}
return nil, err
}
if !resp.Success {
err := fmt.Errorf("sopiga error: %s", resp.Message)
if attempt < w.maxRetries {
waitTime := time.Duration(1<<uint(attempt)) * time.Second
w.logger.Printf("API error, retry in %v", waitTime)
time.Sleep(waitTime)
return w.callCollarAPIWithRetry(ctx, req, attempt+1)
}
return nil, err
}
return resp, nil
}
// callCollarAPI — Single HTTP call
func (w *CollectionBroadcastWorker) callCollarAPI(
ctx context.Context,
req CollarAddRecipientRequest,
) (*CollarAPIResponse, error) {
body, err := json.Marshal(req)
if err != nil {
return nil, fmt.Errorf("json marshal: %w", err)
}
httpReq, err := http.NewRequestWithContext(
ctx,
"POST",
fmt.Sprintf("%s/api/client/collar/add-recipient", w.sopigaBaseURL),
bytes.NewReader(body),
)
if err != nil {
return nil, fmt.Errorf("request creation: %w", err)
}
httpReq.Header.Set("Authorization", fmt.Sprintf("Bearer %s", w.sopigaToken))
httpReq.Header.Set("Content-Type", "application/json")
httpResp, err := w.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 CollarAPIResponse
if err := json.Unmarshal(bodyBytes, &resp); err != nil {
return nil, fmt.Errorf("json unmarshal: %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
}
// updateStatus — Update status ke database
func (w *CollectionBroadcastWorker) updateStatus(
ctx context.Context,
recordID int64,
status string,
errMsg string,
sopigaResp *CollarAPIResponse,
) {
var respJSON []byte
if sopigaResp != nil {
respJSON, _ = json.Marshal(sopigaResp)
}
query := `
SELECT collection_broadcasts.update_broadcast_status($1, $2, $3, $4)
`
_, err := w.db.ExecContext(ctx, query, recordID, status, errMsg, respJSON)
if err != nil {
w.logger.Printf("ERROR updating status for record %d: %v", recordID, err)
}
}
// updateDispatchedStatus — Update ke dispatched + recipient_detail_id
func (w *CollectionBroadcastWorker) updateDispatchedStatus(
ctx context.Context,
recordID int64,
recipientDetailID int64,
) {
query := `
UPDATE collection_broadcasts.broadcast_staging
SET
status = 'dispatched',
sopiga_recipient_detail_id = $1,
dispatched_at = NOW(),
updated_at = NOW()
WHERE id = $2
`
_, err := w.db.ExecContext(ctx, query, recipientDetailID, recordID)
if err != nil {
w.logger.Printf("ERROR updating dispatched status for record %d: %v", recordID, err)
}
}

319
docs/Gadai_collection_broadcast_simplified_schema.sql Normal file
View File

@ -0,0 +1,319 @@
-- ============================================================================
-- GADAI MULIA - COLLECTION BROADCAST TO SOPIGA COLLAR
-- SIMPLIFIED SCHEMA: Denormalized broadcast_staging
-- ============================================================================
CREATE SCHEMA IF NOT EXISTS collection_broadcasts;
-- ============================================================================
-- 1. CONFIGURATION TABLES (Reference only)
-- ============================================================================
-- Sopiga Template Config (pre-setup di Sopiga)
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), -- utility, marketing, notification
channel VARCHAR(50), -- waba
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);
-- Template Variable Mapping (untuk dynamic message building)
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, -- 1, 2, 3, ... (urutan interpolasi)
sopiga_variable_name VARCHAR(100) NOT NULL, -- "Nama", "TotalTagihan", "NoKontrak", etc
variable_type VARCHAR(50), -- string, integer, date, decimal
db_field_source VARCHAR(100) NOT NULL, -- Key di message_payload
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);
-- Sopiga Collar Config (pre-created di Sopiga)
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), -- open, closed
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
-- ============================================================================
-- Single source of truth: message_payload (JSONB)
-- Semua data sudah ada di sini, tidak perlu FK ke nasabah/contract
-- Worker query hanya dari tabel ini
CREATE TABLE collection_broadcasts.broadcast_staging (
id BIGSERIAL PRIMARY KEY,
-- Sopiga config
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),
-- Message payload - SINGLE SOURCE OF TRUTH
-- Contains all fields needed: nasabah_nama, nasabah_phone, nominal_tagihan, tanggal_tempo, invoice_pdf_url, contract_no, etc
-- Struktur dinamis sesuai template variables
-- Example: {
-- "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"
-- }
message_payload JSONB NOT NULL,
-- Status tracking
status VARCHAR(50) DEFAULT 'pending' NOT NULL,
-- pending: waiting to dispatch
-- dispatched: sent to Sopiga queue (HTTP 202 Accepted)
-- delivered: confirmed delivered to WhatsApp
-- failed: error occurred
-- retry_scheduled: queued for retry
error_message TEXT,
error_count INT DEFAULT 0,
-- Sopiga response
sopiga_recipient_detail_id BIGINT,
sopiga_external_id VARCHAR(100),
-- Timestamps
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)
);
-- 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');

453
docs/implementasi_guide.md Normal file
View File

@ -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 ✅

319
docs/quick_start.md Normal file
View File

@ -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=<timestamp>
```
---
## 📊 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! 🚀**

89
docs/webhook_integration_request.md Normal file
View File

@ -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://<DOMAIN_PUBLIK_KAMI>/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: <hex HMAC-SHA256 dari raw body, pakai secret di atas>
```
**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.

63
domain/broadcast.go Normal file
View File

@ -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
}

54
dto/sopiga.go Normal file
View File

@ -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"`
}

9
dto/webhook.go Normal file
View File

@ -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
}

24
entity/broadcast.go Normal file
View File

@ -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"`
}

20
go.mod Normal file
View File

@ -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

28
go.sum Normal file
View File

@ -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=

82
handler/webhook.go Normal file
View File

@ -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
}

83
main.go Normal file
View File

@ -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)
}

17
migrations/001_create_collection_broadcasts.down.sql Normal file
View File

@ -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;

258
migrations/001_create_collection_broadcasts.up.sql Normal file
View File

@ -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;

171
repository/broadcast.go Normal file
View File

@ -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
}

205
service/broadcast.go Normal file
View File

@ -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)
}
}

47
tests/transformer_test.go Normal file
View File

@ -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")
}
}

33
tests/validator_test.go Normal file
View File

@ -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)
}
}

76
transformer/broadcast.go Normal file
View File

@ -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)
}
}

22
validator/broadcast.go Normal file
View File

@ -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
}

85
worker/broadcast_worker.go Normal file
View File

@ -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)
}

83
worker/delivery_sync_worker.go Normal file
View File

@ -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)
}