Rogeeandrogee
d1b11ce3ec
* H-162: confirm Shangwutong sends from echoes * fix(shangwutong): match echoes during send * fix(shangwutong): preserve accepted delivery metric --------- Co-authored-by: Rogee <rogee@ipao.vip>
826 lines
25 KiB
Go
826 lines
25 KiB
Go
// Code generated by sqlc. DO NOT EDIT.
|
|
// versions:
|
|
// sqlc v1.31.1
|
|
// source: outbound.sql
|
|
|
|
package dbgen
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
)
|
|
|
|
const awaitOutboundEcho = `-- name: AwaitOutboundEcho :exec
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'uncertain',
|
|
external_id = NULL,
|
|
external_error_code = NULL,
|
|
status_sync_status = 'not_required',
|
|
claimed_at = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'delivering'
|
|
`
|
|
|
|
func (q *Queries) AwaitOutboundEcho(ctx context.Context, id int64) error {
|
|
_, err := q.db.ExecContext(ctx, awaitOutboundEcho, id)
|
|
return err
|
|
}
|
|
|
|
const awaitOutboundPartEcho = `-- name: AwaitOutboundPartEcho :exec
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'uncertain',
|
|
external_id = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'pending'
|
|
`
|
|
|
|
func (q *Queries) AwaitOutboundPartEcho(ctx context.Context, id int64) error {
|
|
_, err := q.db.ExecContext(ctx, awaitOutboundPartEcho, id)
|
|
return err
|
|
}
|
|
|
|
const claimOutboundMessage = `-- name: ClaimOutboundMessage :one
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'delivering',
|
|
claimed_at = CURRENT_TIMESTAMP,
|
|
attempts = attempts + 1,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = (
|
|
SELECT candidate.id
|
|
FROM outbound_messages AS candidate
|
|
JOIN accounts AS account ON account.id = candidate.account_id
|
|
WHERE candidate.delivery_status = 'pending'
|
|
AND account.enabled = 1
|
|
AND account.desired_presence <> 'offline'
|
|
AND account.deleted_at IS NULL
|
|
AND (candidate.next_attempt_at IS NULL OR julianday(candidate.next_attempt_at) IS NULL OR julianday(candidate.next_attempt_at) <= julianday('now'))
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM outbound_messages AS earlier
|
|
WHERE earlier.account_id = candidate.account_id
|
|
AND earlier.id < candidate.id
|
|
AND earlier.delivery_status IN ('pending', 'delivering', 'uncertain')
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM outbound_operations AS operation
|
|
WHERE operation.account_id = candidate.account_id
|
|
AND operation.delivery_status IN ('delivering', 'uncertain')
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM outbound_operations AS earlier_operation
|
|
WHERE earlier_operation.account_id = candidate.account_id
|
|
AND earlier_operation.delivery_status = 'pending'
|
|
AND (
|
|
julianday(earlier_operation.occurred_at) < julianday(candidate.occurred_at) OR
|
|
(julianday(earlier_operation.occurred_at) = julianday(candidate.occurred_at) AND earlier_operation.event_id < candidate.event_id)
|
|
)
|
|
)
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM outbound_messages AS active
|
|
WHERE active.account_id = candidate.account_id
|
|
AND active.delivery_status = 'delivering'
|
|
)
|
|
ORDER BY candidate.id
|
|
LIMIT 1
|
|
)
|
|
AND delivery_status = 'pending'
|
|
RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
|
|
`
|
|
|
|
func (q *Queries) ClaimOutboundMessage(ctx context.Context) (*OutboundMessage, error) {
|
|
row := q.db.QueryRowContext(ctx, claimOutboundMessage)
|
|
var i OutboundMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AccountID,
|
|
&i.SwtSid,
|
|
&i.EventID,
|
|
&i.OccurredAt,
|
|
&i.GochatMessageID,
|
|
&i.RetryVersion,
|
|
&i.MessageType,
|
|
&i.Content,
|
|
&i.Payload,
|
|
&i.DeliveryStatus,
|
|
&i.ResultVersion,
|
|
&i.ExternalID,
|
|
&i.ExternalErrorCode,
|
|
&i.ClaimedAt,
|
|
&i.Attempts,
|
|
&i.NextAttemptAt,
|
|
&i.StatusSyncStatus,
|
|
&i.StatusSyncAttempts,
|
|
&i.StatusSyncNextAt,
|
|
&i.StatusReportedAt,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return &i, err
|
|
}
|
|
|
|
const claimStatusSync = `-- name: ClaimStatusSync :one
|
|
UPDATE outbound_messages SET
|
|
status_sync_status = 'syncing',
|
|
status_sync_attempts = status_sync_attempts + 1,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = (
|
|
SELECT id FROM outbound_messages
|
|
WHERE status_sync_status = 'pending'
|
|
AND (status_sync_next_at IS NULL OR julianday(status_sync_next_at) IS NULL OR julianday(status_sync_next_at) <= julianday('now'))
|
|
ORDER BY id
|
|
LIMIT 1
|
|
)
|
|
AND status_sync_status = 'pending'
|
|
RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
|
|
`
|
|
|
|
func (q *Queries) ClaimStatusSync(ctx context.Context) (*OutboundMessage, error) {
|
|
row := q.db.QueryRowContext(ctx, claimStatusSync)
|
|
var i OutboundMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AccountID,
|
|
&i.SwtSid,
|
|
&i.EventID,
|
|
&i.OccurredAt,
|
|
&i.GochatMessageID,
|
|
&i.RetryVersion,
|
|
&i.MessageType,
|
|
&i.Content,
|
|
&i.Payload,
|
|
&i.DeliveryStatus,
|
|
&i.ResultVersion,
|
|
&i.ExternalID,
|
|
&i.ExternalErrorCode,
|
|
&i.ClaimedAt,
|
|
&i.Attempts,
|
|
&i.NextAttemptAt,
|
|
&i.StatusSyncStatus,
|
|
&i.StatusSyncAttempts,
|
|
&i.StatusSyncNextAt,
|
|
&i.StatusReportedAt,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return &i, err
|
|
}
|
|
|
|
const completeOutboundMessage = `-- name: CompleteOutboundMessage :exec
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'delivered',
|
|
result_version = result_version + 1,
|
|
external_id = ?,
|
|
external_error_code = NULL,
|
|
status_sync_status = 'pending',
|
|
claimed_at = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'delivering'
|
|
`
|
|
|
|
type CompleteOutboundMessageParams struct {
|
|
ExternalID *string `json:"external_id"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) CompleteOutboundMessage(ctx context.Context, arg CompleteOutboundMessageParams) error {
|
|
_, err := q.db.ExecContext(ctx, completeOutboundMessage, arg.ExternalID, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const completeOutboundPart = `-- name: CompleteOutboundPart :exec
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'delivered',
|
|
external_id = ?,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'pending'
|
|
`
|
|
|
|
type CompleteOutboundPartParams struct {
|
|
ExternalID *string `json:"external_id"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) CompleteOutboundPart(ctx context.Context, arg CompleteOutboundPartParams) error {
|
|
_, err := q.db.ExecContext(ctx, completeOutboundPart, arg.ExternalID, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const completeStatusSync = `-- name: CompleteStatusSync :exec
|
|
UPDATE outbound_messages SET
|
|
status_sync_status = 'synced',
|
|
status_reported_at = CURRENT_TIMESTAMP,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND status_sync_status = 'syncing'
|
|
`
|
|
|
|
func (q *Queries) CompleteStatusSync(ctx context.Context, id int64) error {
|
|
_, err := q.db.ExecContext(ctx, completeStatusSync, id)
|
|
return err
|
|
}
|
|
|
|
const confirmOutboundEcho = `-- name: ConfirmOutboundEcho :execrows
|
|
UPDATE outbound_messages SET
|
|
delivery_status = CASE
|
|
WHEN EXISTS (
|
|
SELECT 1 FROM outbound_parts
|
|
WHERE outbound_message_id = outbound_messages.id
|
|
AND delivery_status <> 'delivered'
|
|
) THEN 'uncertain'
|
|
ELSE 'delivered'
|
|
END,
|
|
result_version = result_version + 1,
|
|
external_id = ?1,
|
|
external_error_code = NULL,
|
|
status_sync_status = CASE
|
|
WHEN EXISTS (
|
|
SELECT 1 FROM outbound_parts
|
|
WHERE outbound_message_id = outbound_messages.id
|
|
AND delivery_status <> 'delivered'
|
|
) THEN 'not_required'
|
|
ELSE 'pending'
|
|
END,
|
|
status_sync_next_at = NULL,
|
|
claimed_at = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE outbound_messages.id = ?2
|
|
AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
|
|
AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout')
|
|
`
|
|
|
|
type ConfirmOutboundEchoParams struct {
|
|
ExternalID *string `json:"external_id"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) ConfirmOutboundEcho(ctx context.Context, arg ConfirmOutboundEchoParams) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, confirmOutboundEcho, arg.ExternalID, arg.ID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const confirmOutboundPartEcho = `-- name: ConfirmOutboundPartEcho :execrows
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'delivered',
|
|
external_id = ?,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
AND external_id IS NULL
|
|
AND delivery_status IN ('pending', 'delivered', 'uncertain')
|
|
`
|
|
|
|
type ConfirmOutboundPartEchoParams struct {
|
|
ExternalID *string `json:"external_id"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) ConfirmOutboundPartEcho(ctx context.Context, arg ConfirmOutboundPartEchoParams) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, confirmOutboundPartEcho, arg.ExternalID, arg.ID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const failExpiredUncertainMessages = `-- name: FailExpiredUncertainMessages :execrows
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'failed',
|
|
result_version = result_version + 1,
|
|
external_error_code = 'uncertain_timeout',
|
|
status_sync_status = 'pending',
|
|
claimed_at = NULL,
|
|
last_error = 'no matching Shangwutong echo arrived before the observation window expired',
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE delivery_status = 'uncertain'
|
|
AND julianday(updated_at) <= julianday(?)
|
|
`
|
|
|
|
func (q *Queries) FailExpiredUncertainMessages(ctx context.Context, julianday interface{}) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, failExpiredUncertainMessages, julianday)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const failOutboundMessage = `-- name: FailOutboundMessage :exec
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'failed',
|
|
result_version = result_version + 1,
|
|
external_error_code = ?,
|
|
status_sync_status = 'pending',
|
|
claimed_at = NULL,
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status IN ('delivering', 'uncertain')
|
|
`
|
|
|
|
type FailOutboundMessageParams struct {
|
|
ExternalErrorCode *string `json:"external_error_code"`
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) FailOutboundMessage(ctx context.Context, arg FailOutboundMessageParams) error {
|
|
_, err := q.db.ExecContext(ctx, failOutboundMessage, arg.ExternalErrorCode, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const failOutboundPart = `-- name: FailOutboundPart :exec
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'failed',
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'pending'
|
|
`
|
|
|
|
type FailOutboundPartParams struct {
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) FailOutboundPart(ctx context.Context, arg FailOutboundPartParams) error {
|
|
_, err := q.db.ExecContext(ctx, failOutboundPart, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const getOutboundByGoChatMessageID = `-- name: GetOutboundByGoChatMessageID :one
|
|
SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at FROM outbound_messages WHERE gochat_message_id = ? LIMIT 1
|
|
`
|
|
|
|
func (q *Queries) GetOutboundByGoChatMessageID(ctx context.Context, gochatMessageID int64) (*OutboundMessage, error) {
|
|
row := q.db.QueryRowContext(ctx, getOutboundByGoChatMessageID, gochatMessageID)
|
|
var i OutboundMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AccountID,
|
|
&i.SwtSid,
|
|
&i.EventID,
|
|
&i.OccurredAt,
|
|
&i.GochatMessageID,
|
|
&i.RetryVersion,
|
|
&i.MessageType,
|
|
&i.Content,
|
|
&i.Payload,
|
|
&i.DeliveryStatus,
|
|
&i.ResultVersion,
|
|
&i.ExternalID,
|
|
&i.ExternalErrorCode,
|
|
&i.ClaimedAt,
|
|
&i.Attempts,
|
|
&i.NextAttemptAt,
|
|
&i.StatusSyncStatus,
|
|
&i.StatusSyncAttempts,
|
|
&i.StatusSyncNextAt,
|
|
&i.StatusReportedAt,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return &i, err
|
|
}
|
|
|
|
const insertOutboundMessage = `-- name: InsertOutboundMessage :one
|
|
INSERT INTO outbound_messages (
|
|
account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version,
|
|
message_type, content, payload
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(gochat_message_id) DO NOTHING
|
|
RETURNING id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at
|
|
`
|
|
|
|
type InsertOutboundMessageParams struct {
|
|
AccountID int64 `json:"account_id"`
|
|
SwtSid string `json:"swt_sid"`
|
|
EventID string `json:"event_id"`
|
|
OccurredAt time.Time `json:"occurred_at"`
|
|
GochatMessageID int64 `json:"gochat_message_id"`
|
|
RetryVersion int64 `json:"retry_version"`
|
|
MessageType string `json:"message_type"`
|
|
Content *string `json:"content"`
|
|
Payload string `json:"payload"`
|
|
}
|
|
|
|
func (q *Queries) InsertOutboundMessage(ctx context.Context, arg InsertOutboundMessageParams) (*OutboundMessage, error) {
|
|
row := q.db.QueryRowContext(ctx, insertOutboundMessage,
|
|
arg.AccountID,
|
|
arg.SwtSid,
|
|
arg.EventID,
|
|
arg.OccurredAt,
|
|
arg.GochatMessageID,
|
|
arg.RetryVersion,
|
|
arg.MessageType,
|
|
arg.Content,
|
|
arg.Payload,
|
|
)
|
|
var i OutboundMessage
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.AccountID,
|
|
&i.SwtSid,
|
|
&i.EventID,
|
|
&i.OccurredAt,
|
|
&i.GochatMessageID,
|
|
&i.RetryVersion,
|
|
&i.MessageType,
|
|
&i.Content,
|
|
&i.Payload,
|
|
&i.DeliveryStatus,
|
|
&i.ResultVersion,
|
|
&i.ExternalID,
|
|
&i.ExternalErrorCode,
|
|
&i.ClaimedAt,
|
|
&i.Attempts,
|
|
&i.NextAttemptAt,
|
|
&i.StatusSyncStatus,
|
|
&i.StatusSyncAttempts,
|
|
&i.StatusSyncNextAt,
|
|
&i.StatusReportedAt,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return &i, err
|
|
}
|
|
|
|
const insertOutboundPart = `-- name: InsertOutboundPart :one
|
|
INSERT INTO outbound_parts (
|
|
outbound_message_id, part_index, part_type, attachment_id,
|
|
content, data_url, file_name, file_size, voice
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
RETURNING id, outbound_message_id, part_index, part_type, attachment_id, content, data_url, file_name, file_size, voice, delivery_status, external_id, last_error, created_at, updated_at
|
|
`
|
|
|
|
type InsertOutboundPartParams struct {
|
|
OutboundMessageID int64 `json:"outbound_message_id"`
|
|
PartIndex int64 `json:"part_index"`
|
|
PartType string `json:"part_type"`
|
|
AttachmentID *int64 `json:"attachment_id"`
|
|
Content *string `json:"content"`
|
|
DataUrl *string `json:"data_url"`
|
|
FileName *string `json:"file_name"`
|
|
FileSize *int64 `json:"file_size"`
|
|
Voice int64 `json:"voice"`
|
|
}
|
|
|
|
func (q *Queries) InsertOutboundPart(ctx context.Context, arg InsertOutboundPartParams) (*OutboundPart, error) {
|
|
row := q.db.QueryRowContext(ctx, insertOutboundPart,
|
|
arg.OutboundMessageID,
|
|
arg.PartIndex,
|
|
arg.PartType,
|
|
arg.AttachmentID,
|
|
arg.Content,
|
|
arg.DataUrl,
|
|
arg.FileName,
|
|
arg.FileSize,
|
|
arg.Voice,
|
|
)
|
|
var i OutboundPart
|
|
err := row.Scan(
|
|
&i.ID,
|
|
&i.OutboundMessageID,
|
|
&i.PartIndex,
|
|
&i.PartType,
|
|
&i.AttachmentID,
|
|
&i.Content,
|
|
&i.DataUrl,
|
|
&i.FileName,
|
|
&i.FileSize,
|
|
&i.Voice,
|
|
&i.DeliveryStatus,
|
|
&i.ExternalID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
)
|
|
return &i, err
|
|
}
|
|
|
|
const listOutboundEchoCandidates = `-- name: ListOutboundEchoCandidates :many
|
|
SELECT id, account_id, swt_sid, event_id, occurred_at, gochat_message_id, retry_version, message_type, content, payload, delivery_status, result_version, external_id, external_error_code, claimed_at, attempts, next_attempt_at, status_sync_status, status_sync_attempts, status_sync_next_at, status_reported_at, last_error, created_at, updated_at FROM outbound_messages
|
|
WHERE account_id = ?
|
|
AND swt_sid = ?
|
|
AND delivery_status IN ('delivering', 'delivered', 'uncertain', 'failed')
|
|
AND (delivery_status <> 'failed' OR external_error_code = 'uncertain_timeout')
|
|
AND EXISTS (
|
|
SELECT 1 FROM outbound_parts
|
|
WHERE outbound_message_id = outbound_messages.id
|
|
AND external_id IS NULL
|
|
AND delivery_status IN ('pending', 'delivered', 'uncertain')
|
|
)
|
|
ORDER BY id
|
|
LIMIT 20
|
|
`
|
|
|
|
type ListOutboundEchoCandidatesParams struct {
|
|
AccountID int64 `json:"account_id"`
|
|
SwtSid string `json:"swt_sid"`
|
|
}
|
|
|
|
func (q *Queries) ListOutboundEchoCandidates(ctx context.Context, arg ListOutboundEchoCandidatesParams) ([]*OutboundMessage, error) {
|
|
rows, err := q.db.QueryContext(ctx, listOutboundEchoCandidates, arg.AccountID, arg.SwtSid)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []*OutboundMessage{}
|
|
for rows.Next() {
|
|
var i OutboundMessage
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.AccountID,
|
|
&i.SwtSid,
|
|
&i.EventID,
|
|
&i.OccurredAt,
|
|
&i.GochatMessageID,
|
|
&i.RetryVersion,
|
|
&i.MessageType,
|
|
&i.Content,
|
|
&i.Payload,
|
|
&i.DeliveryStatus,
|
|
&i.ResultVersion,
|
|
&i.ExternalID,
|
|
&i.ExternalErrorCode,
|
|
&i.ClaimedAt,
|
|
&i.Attempts,
|
|
&i.NextAttemptAt,
|
|
&i.StatusSyncStatus,
|
|
&i.StatusSyncAttempts,
|
|
&i.StatusSyncNextAt,
|
|
&i.StatusReportedAt,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, &i)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const listOutboundParts = `-- name: ListOutboundParts :many
|
|
SELECT id, outbound_message_id, part_index, part_type, attachment_id, content, data_url, file_name, file_size, voice, delivery_status, external_id, last_error, created_at, updated_at FROM outbound_parts
|
|
WHERE outbound_message_id = ?
|
|
ORDER BY part_index
|
|
`
|
|
|
|
func (q *Queries) ListOutboundParts(ctx context.Context, outboundMessageID int64) ([]*OutboundPart, error) {
|
|
rows, err := q.db.QueryContext(ctx, listOutboundParts, outboundMessageID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
items := []*OutboundPart{}
|
|
for rows.Next() {
|
|
var i OutboundPart
|
|
if err := rows.Scan(
|
|
&i.ID,
|
|
&i.OutboundMessageID,
|
|
&i.PartIndex,
|
|
&i.PartType,
|
|
&i.AttachmentID,
|
|
&i.Content,
|
|
&i.DataUrl,
|
|
&i.FileName,
|
|
&i.FileSize,
|
|
&i.Voice,
|
|
&i.DeliveryStatus,
|
|
&i.ExternalID,
|
|
&i.LastError,
|
|
&i.CreatedAt,
|
|
&i.UpdatedAt,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
items = append(items, &i)
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return nil, err
|
|
}
|
|
return items, nil
|
|
}
|
|
|
|
const markOutboundPartUncertain = `-- name: MarkOutboundPartUncertain :exec
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'uncertain',
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'pending'
|
|
`
|
|
|
|
type MarkOutboundPartUncertainParams struct {
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) MarkOutboundPartUncertain(ctx context.Context, arg MarkOutboundPartUncertainParams) error {
|
|
_, err := q.db.ExecContext(ctx, markOutboundPartUncertain, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const markOutboundUncertain = `-- name: MarkOutboundUncertain :exec
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'uncertain',
|
|
result_version = result_version + 1,
|
|
external_error_code = ?,
|
|
status_sync_status = 'pending',
|
|
claimed_at = NULL,
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'delivering'
|
|
`
|
|
|
|
type MarkOutboundUncertainParams struct {
|
|
ExternalErrorCode *string `json:"external_error_code"`
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) MarkOutboundUncertain(ctx context.Context, arg MarkOutboundUncertainParams) error {
|
|
_, err := q.db.ExecContext(ctx, markOutboundUncertain, arg.ExternalErrorCode, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const recoverOutboundDeliveriesAsUncertain = `-- name: RecoverOutboundDeliveriesAsUncertain :execrows
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'uncertain',
|
|
result_version = result_version + 1,
|
|
external_error_code = 'connector_restart_uncertain',
|
|
status_sync_status = 'pending',
|
|
claimed_at = NULL,
|
|
last_error = 'connector restarted while request was delivering',
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE delivery_status = 'delivering'
|
|
`
|
|
|
|
func (q *Queries) RecoverOutboundDeliveriesAsUncertain(ctx context.Context) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, recoverOutboundDeliveriesAsUncertain)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const recoverOutboundPartDeliveriesAsUncertain = `-- name: RecoverOutboundPartDeliveriesAsUncertain :execrows
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'uncertain',
|
|
last_error = 'connector restarted while part request may have been in flight',
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id IN (
|
|
SELECT MIN(candidate.id)
|
|
FROM outbound_parts AS candidate
|
|
JOIN outbound_messages AS message ON message.id = candidate.outbound_message_id
|
|
WHERE message.delivery_status = 'delivering'
|
|
AND candidate.delivery_status = 'pending'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM outbound_parts AS uncertain
|
|
WHERE uncertain.outbound_message_id = candidate.outbound_message_id
|
|
AND uncertain.delivery_status = 'uncertain'
|
|
)
|
|
GROUP BY candidate.outbound_message_id
|
|
)
|
|
`
|
|
|
|
func (q *Queries) RecoverOutboundPartDeliveriesAsUncertain(ctx context.Context) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, recoverOutboundPartDeliveriesAsUncertain)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const recoverOutboundStatusSyncs = `-- name: RecoverOutboundStatusSyncs :execrows
|
|
UPDATE outbound_messages SET
|
|
status_sync_status = 'pending',
|
|
status_sync_next_at = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE status_sync_status = 'syncing'
|
|
`
|
|
|
|
func (q *Queries) RecoverOutboundStatusSyncs(ctx context.Context) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, recoverOutboundStatusSyncs)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const resetUndeliveredOutboundParts = `-- name: ResetUndeliveredOutboundParts :exec
|
|
UPDATE outbound_parts SET
|
|
delivery_status = 'pending',
|
|
external_id = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE outbound_message_id = ?
|
|
AND delivery_status IN ('failed', 'uncertain')
|
|
`
|
|
|
|
func (q *Queries) ResetUndeliveredOutboundParts(ctx context.Context, outboundMessageID int64) error {
|
|
_, err := q.db.ExecContext(ctx, resetUndeliveredOutboundParts, outboundMessageID)
|
|
return err
|
|
}
|
|
|
|
const retryOutboundDelivery = `-- name: RetryOutboundDelivery :exec
|
|
UPDATE outbound_messages SET
|
|
delivery_status = 'pending',
|
|
claimed_at = NULL,
|
|
next_attempt_at = ?,
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND delivery_status = 'delivering'
|
|
`
|
|
|
|
type RetryOutboundDeliveryParams struct {
|
|
NextAttemptAt *time.Time `json:"next_attempt_at"`
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) RetryOutboundDelivery(ctx context.Context, arg RetryOutboundDeliveryParams) error {
|
|
_, err := q.db.ExecContext(ctx, retryOutboundDelivery, arg.NextAttemptAt, arg.LastError, arg.ID)
|
|
return err
|
|
}
|
|
|
|
const retryOutboundMessage = `-- name: RetryOutboundMessage :execrows
|
|
UPDATE outbound_messages SET
|
|
event_id = ?1,
|
|
occurred_at = ?2,
|
|
retry_version = ?3,
|
|
message_type = ?4,
|
|
content = ?5,
|
|
payload = ?6,
|
|
delivery_status = 'pending',
|
|
status_sync_status = 'not_required',
|
|
next_attempt_at = NULL,
|
|
claimed_at = NULL,
|
|
last_error = NULL,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE gochat_message_id = ?7
|
|
AND delivery_status = 'failed'
|
|
AND retry_version < ?3
|
|
`
|
|
|
|
type RetryOutboundMessageParams struct {
|
|
EventID string `json:"event_id"`
|
|
OccurredAt time.Time `json:"occurred_at"`
|
|
RetryVersion int64 `json:"retry_version"`
|
|
MessageType string `json:"message_type"`
|
|
Content *string `json:"content"`
|
|
Payload string `json:"payload"`
|
|
GochatMessageID int64 `json:"gochat_message_id"`
|
|
}
|
|
|
|
func (q *Queries) RetryOutboundMessage(ctx context.Context, arg RetryOutboundMessageParams) (int64, error) {
|
|
result, err := q.db.ExecContext(ctx, retryOutboundMessage,
|
|
arg.EventID,
|
|
arg.OccurredAt,
|
|
arg.RetryVersion,
|
|
arg.MessageType,
|
|
arg.Content,
|
|
arg.Payload,
|
|
arg.GochatMessageID,
|
|
)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return result.RowsAffected()
|
|
}
|
|
|
|
const retryStatusSync = `-- name: RetryStatusSync :exec
|
|
UPDATE outbound_messages SET
|
|
status_sync_status = 'pending',
|
|
status_sync_next_at = ?,
|
|
last_error = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ? AND status_sync_status = 'syncing'
|
|
`
|
|
|
|
type RetryStatusSyncParams struct {
|
|
StatusSyncNextAt *time.Time `json:"status_sync_next_at"`
|
|
LastError *string `json:"last_error"`
|
|
ID int64 `json:"id"`
|
|
}
|
|
|
|
func (q *Queries) RetryStatusSync(ctx context.Context, arg RetryStatusSyncParams) error {
|
|
_, err := q.db.ExecContext(ctx, retryStatusSync, arg.StatusSyncNextAt, arg.LastError, arg.ID)
|
|
return err
|
|
}
|