omnix-sopiga/docs/Collection_broadcast_worker_simplified.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)
	}
}