Open Source Release
225 lines
7.2 KiB
Go
225 lines
7.2 KiB
Go
// Package clientnotify sends client-facing Telegram/MAX/SMS notifications —
|
|
// order created, status changed, ready for pickup, warranty expiring,
|
|
// loyalty points accrued. Distinct from internal/notify (a single fixed
|
|
// staff chat, fire-and-forget): every client is a different recipient with
|
|
// their own channel preference, so delivery goes through an outbox table
|
|
// (client_notifications) with per-event deduplication instead.
|
|
//
|
|
// Telegram's bot token and API base URL are shared with internal/notify
|
|
// (one bot, read from the same internal/settings row) — only the recipient
|
|
// chat_id differs. MAX is a separate bot/token (see internal/maxbot for the
|
|
// inbound linking side). SMS goes through internal/smsgw, configured
|
|
// separately since a deployment might enable Telegram/MAX notifications
|
|
// without ever setting up an SMS aggregator.
|
|
package clientnotify
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"log"
|
|
"net/http"
|
|
"time"
|
|
|
|
"production/internal/dbutil"
|
|
"production/internal/settings"
|
|
"production/internal/smsgw"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgxpool"
|
|
)
|
|
|
|
const sendTimeout = 10 * time.Second
|
|
|
|
type Handler struct {
|
|
db *pgxpool.Pool
|
|
client *http.Client
|
|
}
|
|
|
|
func NewHandler(db *pgxpool.Pool) *Handler {
|
|
return &Handler{db: db, client: &http.Client{Timeout: sendTimeout}}
|
|
}
|
|
|
|
// Event describes one notification-worthy moment. TGBody/SMSBody are
|
|
// pre-rendered by the caller via templates.go (only the caller — the order
|
|
// or cartridge handler — knows the domain specifics like device label or
|
|
// status text); Enqueue only decides whether/how to deliver them. MAX gets
|
|
// no body field of its own — every templates.go function's "rich text"
|
|
// branch (anything that isn't the SMS branch) reads fine as a MAX message
|
|
// too, so the max channel just reuses TGBody rather than the caller
|
|
// rendering a third, identical variant.
|
|
type Event struct {
|
|
ClientID string
|
|
OrderID string
|
|
CartridgeBatchID string
|
|
Trigger string
|
|
// DedupeSeed identifies the specific occurrence that triggered this
|
|
// notification — an order_events.id for status changes, a composite
|
|
// like "warranty:<order_id>:<date>" for scheduled reminders. See
|
|
// dedupeKey's doc comment.
|
|
DedupeSeed string
|
|
TGBody string
|
|
SMSBody string
|
|
}
|
|
|
|
// Enqueue looks up the client's channel preference, picks Telegram or SMS,
|
|
// writes an idempotent outbox row, and dispatches it in the background —
|
|
// fire-and-forget from the caller's perspective, same convention as
|
|
// internal/notify.Send. A client with no linked Telegram chat and no SMS
|
|
// opt-in is a silent no-op, not an error.
|
|
func (h *Handler) Enqueue(ev Event) {
|
|
go func() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), sendTimeout)
|
|
defer cancel()
|
|
|
|
s, err := settings.Fetch(ctx, h.db)
|
|
if err != nil {
|
|
log.Printf("clientnotify: settings fetch failed: %v", err)
|
|
return
|
|
}
|
|
if !s.ClientNotifyEnabled {
|
|
return
|
|
}
|
|
|
|
var tgChatID, maxChatID, vkChatID, phone string
|
|
var notifyTelegram, notifyMax, notifyVk, notifySMS bool
|
|
err = h.db.QueryRow(ctx,
|
|
`SELECT COALESCE(tg_chat_id, ''), COALESCE(max_chat_id, ''), COALESCE(vk_chat_id, ''), COALESCE(phone_normalized, ''),
|
|
notify_telegram, notify_max, notify_vk, notify_sms
|
|
FROM clients WHERE id = $1::uuid`,
|
|
ev.ClientID,
|
|
).Scan(&tgChatID, &maxChatID, &vkChatID, &phone, ¬ifyTelegram, ¬ifyMax, ¬ifyVk, ¬ifySMS)
|
|
if err != nil {
|
|
log.Printf("clientnotify: client fetch failed for %s: %v", ev.ClientID, err)
|
|
return
|
|
}
|
|
|
|
channel, ok := chooseChannel(tgChatID, maxChatID, vkChatID, notifyTelegram, notifyMax, notifyVk, notifySMS, s.ClientNotifyEnabled, s.SMSEnabled)
|
|
if !ok {
|
|
return
|
|
}
|
|
|
|
body := ev.TGBody
|
|
if channel == "sms" {
|
|
body = ev.SMSBody
|
|
}
|
|
if body == "" {
|
|
return
|
|
}
|
|
|
|
key := dedupeKey(ev.Trigger, ev.DedupeSeed, channel)
|
|
|
|
var id string
|
|
err = h.db.QueryRow(ctx,
|
|
`INSERT INTO client_notifications (client_id, order_id, cartridge_batch_id, trigger, channel, body, dedupe_key)
|
|
VALUES ($1::uuid, $2::uuid, $3::uuid, $4, $5, $6, $7)
|
|
ON CONFLICT (dedupe_key) DO NOTHING
|
|
RETURNING id`,
|
|
ev.ClientID, dbutil.NullIfEmpty(ev.OrderID), dbutil.NullIfEmpty(ev.CartridgeBatchID), ev.Trigger, channel, body, key,
|
|
).Scan(&id)
|
|
if err != nil {
|
|
if err == pgx.ErrNoRows {
|
|
// Already sent for this exact event — not an error, the
|
|
// unique index on dedupe_key did its job.
|
|
return
|
|
}
|
|
log.Printf("clientnotify: enqueue insert failed: %v", err)
|
|
return
|
|
}
|
|
|
|
h.dispatch(ctx, id, channel, tgChatID, maxChatID, vkChatID, phone, body, s)
|
|
}()
|
|
}
|
|
|
|
func (h *Handler) dispatch(ctx context.Context, id, channel, tgChatID, maxChatID, vkChatID, phone, body string, s settings.Settings) {
|
|
var (
|
|
msgID string
|
|
err error
|
|
)
|
|
switch channel {
|
|
case "telegram":
|
|
msgID, err = h.sendTelegram(ctx, tgChatID, body, s)
|
|
case "max":
|
|
msgID, err = h.sendMax(ctx, maxChatID, body, s)
|
|
case "vk":
|
|
msgID, err = h.sendVk(ctx, vkChatID, body, s)
|
|
case "sms":
|
|
msgID, err = h.sendSMS(ctx, phone, body, s)
|
|
}
|
|
if err != nil {
|
|
h.markFailed(ctx, id, err)
|
|
return
|
|
}
|
|
h.markSent(ctx, id, msgID)
|
|
}
|
|
|
|
func (h *Handler) sendTelegram(ctx context.Context, chatID, text string, s settings.Settings) (string, error) {
|
|
req, err := buildTGSendRequest(ctx, text, s.TGBotToken, chatID, s.TGAPIBaseURL)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
resp, err := h.client.Do(req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer resp.Body.Close()
|
|
return parseTGSendResponse(resp.Body)
|
|
}
|
|
|
|
func (h *Handler) sendMax(ctx context.Context, chatID, text string, s settings.Settings) (string, error) {
|
|
req, err := buildMaxSendRequest(ctx, text, s.MaxBotToken, chatID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
resp, err := h.client.Do(req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer resp.Body.Close()
|
|
return parseMaxSendResponse(resp.StatusCode, resp.Body)
|
|
}
|
|
|
|
func (h *Handler) sendVk(ctx context.Context, chatID, text string, s settings.Settings) (string, error) {
|
|
req, err := buildVkSendRequest(ctx, text, s.VkGroupToken, chatID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
resp, err := h.client.Do(req)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer resp.Body.Close()
|
|
respBody, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return parseVkSendResponse(respBody)
|
|
}
|
|
|
|
func (h *Handler) sendSMS(ctx context.Context, phone, text string, s settings.Settings) (string, error) {
|
|
provider, err := smsgw.New(s.SMSProvider, s.SMSAPIID, s.SMSFrom, false)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
return provider.Send(ctx, phone, text)
|
|
}
|
|
|
|
func (h *Handler) markSent(ctx context.Context, id, msgID string) {
|
|
if _, err := h.db.Exec(ctx,
|
|
`UPDATE client_notifications SET status = 'sent', sent_at = NOW(),
|
|
provider_message_id = $1, attempts = attempts + 1 WHERE id = $2::uuid`,
|
|
dbutil.NullIfEmpty(msgID), id,
|
|
); err != nil {
|
|
log.Printf("clientnotify: mark sent failed for %s: %v", id, err)
|
|
}
|
|
}
|
|
|
|
func (h *Handler) markFailed(ctx context.Context, id string, sendErr error) {
|
|
if _, err := h.db.Exec(ctx,
|
|
`UPDATE client_notifications SET status = 'failed', error = $1, attempts = attempts + 1 WHERE id = $2::uuid`,
|
|
sendErr.Error(), id,
|
|
); err != nil {
|
|
log.Printf("clientnotify: mark failed failed for %s: %v", id, err)
|
|
}
|
|
log.Printf("clientnotify: delivery failed for %s: %v", id, sendErr)
|
|
}
|