452 lines
12 KiB
Go
452 lines
12 KiB
Go
|
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)
|
||
|
|
}
|
||
|
|
}
|