104 lines
3.5 KiB
Go
104 lines
3.5 KiB
Go
package store
|
|
|
|
import (
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"time"
|
|
|
|
"git.ipao.vip/rogee/go-sip/internal/contract"
|
|
"git.ipao.vip/rogee/go-sip/internal/mq"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
)
|
|
|
|
var ErrUploadMismatch = errors.New("upload notification does not match its persisted identity or object")
|
|
|
|
type uploadNotice struct {
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
DispatcherID string `json:"dispatcher_id"`
|
|
TenantID string `json:"tenant_id"`
|
|
TenantKey string `json:"tenant_key"`
|
|
AggregateID string `json:"aggregate_id"`
|
|
Payload map[string]any `json:"payload"`
|
|
}
|
|
|
|
// RecordUploadNotification commits the reported upload fact and its original
|
|
// notification together. It does not mark delivery complete before publication.
|
|
func (s *Store) RecordUploadNotification(uploadID, eventID, tenantKey, routingKey string, body []byte, uploadedAt time.Time) error {
|
|
if uploadID == "" || eventID == "" || tenantKey == "" || routingKey == "" {
|
|
return errors.New("upload notification identity and route are required")
|
|
}
|
|
if err := contract.ValidateMQMessage(body); err != nil {
|
|
return err
|
|
}
|
|
var notice uploadNotice
|
|
if err := json.Unmarshal(body, ¬ice); err != nil {
|
|
return err
|
|
}
|
|
if notice.EventID != eventID || notice.EventType != "recording.uploaded" || notice.TenantKey != tenantKey || notice.Payload["upload_id"] != uploadID {
|
|
return ErrUploadMismatch
|
|
}
|
|
route, err := tenant.NewDispatcherRoute(notice.DispatcherID, tenantKey)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if route.OutboundKey != routingKey {
|
|
return ErrMessageScope
|
|
}
|
|
if uploadedAt.IsZero() {
|
|
return errors.New("upload fact time is required")
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
tx, err := s.db.Begin()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
if err := bindMQScope(tx, notice.DispatcherID, notice.TenantID, tenantKey); err != nil {
|
|
return err
|
|
}
|
|
var existing string
|
|
err = tx.QueryRow(`SELECT event_id FROM upload_notifications WHERE upload_id=?`, uploadID).Scan(&existing)
|
|
if err == nil {
|
|
if existing != eventID {
|
|
return ErrIdempotencyConflict
|
|
}
|
|
var previousBody []byte
|
|
if err := tx.QueryRow(`SELECT body FROM outbox WHERE event_id=?`, existing).Scan(&previousBody); err != nil {
|
|
return err
|
|
}
|
|
var previous uploadNotice
|
|
if err := json.Unmarshal(previousBody, &previous); err != nil {
|
|
return err
|
|
}
|
|
if !reflect.DeepEqual(previous, notice) {
|
|
return ErrIdempotencyConflict
|
|
}
|
|
return nil
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return err
|
|
}
|
|
var state, objectKey, bucket string
|
|
if err := tx.QueryRow(`SELECT state,object_key,d.bucket FROM uploads u JOIN upload_destinations d ON d.upload_id=u.upload_id WHERE u.upload_id=?`, uploadID).Scan(&state, &objectKey, &bucket); err != nil {
|
|
return err
|
|
}
|
|
if notice.Payload["object_key"] != objectKey || notice.Payload["bucket"] != bucket {
|
|
return ErrUploadMismatch
|
|
}
|
|
if state != "granted" {
|
|
return fmt.Errorf("upload cannot report new facts in state %q", state)
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO outbox(event_id,tenant_key,exchange,routing_key,body,status,created_at) VALUES(?,?,?,?,?,'pending',?)`, eventID, tenantKey, mq.EventExchange, routingKey, body, uploadedAt.UTC().Format(time.RFC3339Nano)); err != nil {
|
|
return err
|
|
}
|
|
if _, err := tx.Exec(`INSERT INTO upload_notifications(upload_id,event_id,uploaded_at) VALUES(?,?,?)`, uploadID, eventID, uploadedAt.UTC().Format(time.RFC3339Nano)); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|