rogee/gochat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

H-47: retry CID until contact exists (#13)

Browse Source 
Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
2026-08-13 19:57:05 +08:00
committed by GitHub
co-authored by rogee
parent d00ea39f6c
commit de285b4f5d
8 changed files with 105 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
channels/shangwutong/db/generated/inbound.sql.go
+1
View File
@@ -26,6 +26,7 @@ WHERE id = (
AND earlier.swt_sid = candidate.swt_sid
AND earlier.id < candidate.id
AND earlier.delivery_status IN ('pending', 'delivering')
AND NOT (earlier.kind IN (24, 52) AND candidate.kind NOT IN (24, 52))
)
ORDER BY candidate.id
LIMIT 1
channels/shangwutong/db/migrations/003_retry_deferred_contact_metadata.down.sql
+1
View File
@@ -0,0 +1 @@
-- Data recovery migration: restored queue state is intentionally not reverted.
channels/shangwutong/db/migrations/003_retry_deferred_contact_metadata.up.sql
+8
View File
@@ -0,0 +1,8 @@
UPDATE inbound_events SET
delivery_status = 'pending',
attempts = 0,
next_attempt_at = NULL,
updated_at = CURRENT_TIMESTAMP
WHERE kind IN (24, 52)
AND delivery_status = 'failed'
AND last_error = 'GoChat API 404 not_found: contact source not found';
channels/shangwutong/db/queries/inbound.sql
+1
View File
@@ -23,6 +23,7 @@ WHERE id = (
AND earlier.swt_sid = candidate.swt_sid
AND earlier.id < candidate.id
AND earlier.delivery_status IN ('pending', 'delivering')
AND NOT (earlier.kind IN (24, 52) AND candidate.kind NOT IN (24, 52))
)
ORDER BY candidate.id
LIMIT 1
channels/shangwutong/internal/command/root_test.go
+1 -1
View File
@@ -87,7 +87,7 @@ func TestOperationalCommands(t *testing.T) {
t.Fatalf("%v produced no output", args)
}
}
if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 2 {
if version, err := store.InspectDatabase(context.Background(), backupPath); err != nil || version != 3 {
t.Fatalf("backup version = %d, %v", version, err)
}
channels/shangwutong/internal/delivery/inbound.go
+3 -2
View File
@@ -475,8 +475,9 @@ func (i *Inbound) complete(event *dbgen.InboundEvent, mapped mappedEvent, messag
func (i *Inbound) retryOrFail(event *dbgen.InboundEvent, mapped mappedEvent, deliveryErr error) error {
var apiErr *gochat.APIError
retryable := !errors.As(deliveryErr, &apiErr) || apiErr.Retryable
if retryable && event.Attempts < 10 {
waitingForContact := mapped.ContactCID != "" && isNotFound(deliveryErr)
retryable := !errors.As(deliveryErr, &apiErr) || apiErr.Retryable || waitingForContact
if retryable && (event.Attempts < 10 || waitingForContact) {
detail := deliveryErr.Error()
next := time.Now().Add(backoffForError(event.Attempts, 5*time.Minute, deliveryErr))
persistErr := i.store.Writer().RetryInboundEvent(context.Background(), dbgen.RetryInboundEventParams{
channels/shangwutong/internal/delivery/inbound_test.go
+40 -1
View File
@@ -93,6 +93,43 @@ func TestInboundKind24PersistsCIDWithoutCreatingResources(t *testing.T) {
}
}
func TestInboundKind24WaitsForLaterContactInboxAndConverges(t *testing.T) {
ctx := context.Background()
database, account := deliveryDatabase(t, ctx)
persistInboundEvent(t, database, account, swt.HeartbeatEvent{
SessionID: "visitor", Kind: 24, Text: "cookie/123", SeqID: 42,
Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 24 [REDACTED] 42 timestamp",
})
client := &inboundRecorder{metadataErr: &gochat.APIError{StatusCode: 404, Code: "not_found", Message: "contact source not found"}}
worker, _ := NewInbound(database, client, nil, 1)
if worked, err := worker.processInbound(ctx); !worked || err == nil {
t.Fatalf("process CID before contact = %v, %v", worked, err)
}
cidEvent, err := database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:24:42")
if err != nil || cidEvent.DeliveryStatus != "pending" || cidEvent.NextAttemptAt == nil || cidEvent.LastError == nil {
t.Fatalf("deferred CID event = %#v, %v", cidEvent, err)
}
persistInboundEvent(t, database, account, swt.HeartbeatEvent{
SessionID: "visitor", Kind: 2, Text: "hello", SeqID: 43,
Timestamp: time.Now().Format(time.RFC3339Nano), RawLine: "visitor 2 hello 43 timestamp",
})
if worked, err := worker.processInbound(ctx); !worked || err != nil {
t.Fatalf("create later contact inbox = %v, %v", worked, err)
}
if delay := time.Until(*cidEvent.NextAttemptAt); delay > 0 {
time.Sleep(delay + 20*time.Millisecond)
}
if worked, err := worker.processInbound(ctx); !worked || err != nil {
t.Fatalf("retry CID after contact = %v, %v", worked, err)
}
cidEvent, err = database.Reader().GetInboundEventByKey(ctx, "swt:10:visitor:24:42")
if err != nil || cidEvent.DeliveryStatus != "delivered" || cidEvent.Attempts != 2 || client.lastSourceID != "visitor" || client.lastCID != "cookie/123" {
t.Fatalf("converged CID event = %#v, client = %#v, err = %v", cidEvent, client, err)
}
}
func TestInboundKind3ConfirmsUniqueOutboundEcho(t *testing.T) {
ctx := context.Background()
database, account := deliveryDatabase(t, ctx)
@@ -457,9 +494,11 @@ type inboundRecorder struct {
lastInboxID int64
lastSourceID string
lastCID string
metadataErr error
}
func (r *inboundRecorder) EnsureContact(_ context.Context, _, _ string, request gochat.ContactRequest) (gochat.Contact, error) {
r.metadataErr = nil
return gochat.Contact{ID: 456, SourceID: request.SourceID, Name: request.Name}, nil
}
@@ -470,7 +509,7 @@ func (r *inboundRecorder) UpdateContact(_ context.Context, _, _, sourceID string
func (r *inboundRecorder) UpdateContactChannelMetadata(_ context.Context, inboxID int64, sourceID, cid string) error {
r.metadataUpdates++
r.lastInboxID, r.lastSourceID, r.lastCID = inboxID, sourceID, cid
return nil
return r.metadataErr
}
func (r *inboundRecorder) EnsureConversation(context.Context, string, string, map[string]any) (gochat.PublicConversation, error) {
channels/shangwutong/internal/store/store_test.go
+50 -1
View File
@@ -102,6 +102,55 @@ func TestMigrationVersionParser(t *testing.T) {
}
}
func TestDeferredContactMetadataMigrationOnlyRetriesMatchingFailures(t *testing.T) {
db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "migration.db"))
if err != nil {
t.Fatal(err)
}
defer db.Close()
if _, err := db.Exec(`CREATE TABLE inbound_events (
id INTEGER PRIMARY KEY, kind INTEGER, delivery_status TEXT, attempts INTEGER,
next_attempt_at DATETIME, last_error TEXT, updated_at DATETIME)`); err != nil {
t.Fatal(err)
}
for _, values := range []string{
`(1,24,'failed',10,CURRENT_TIMESTAMP,'GoChat API 404 not_found: contact source not found',CURRENT_TIMESTAMP)`,
`(2,24,'failed',10,CURRENT_TIMESTAMP,'other failure',CURRENT_TIMESTAMP)`,
`(3,2,'failed',10,CURRENT_TIMESTAMP,'GoChat API 404 not_found: contact source not found',CURRENT_TIMESTAMP)`,
} {
if _, err := db.Exec(`INSERT INTO inbound_events VALUES ` + values); err != nil {
t.Fatal(err)
}
}
migration, err := os.ReadFile(filepath.Join("..", "..", "db", "migrations", "003_retry_deferred_contact_metadata.up.sql"))
if err != nil {
t.Fatal(err)
}
if _, err := db.Exec(string(migration)); err != nil {
t.Fatal(err)
}
rows, err := db.Query(`SELECT id, delivery_status, attempts, next_attempt_at FROM inbound_events ORDER BY id`)
if err != nil {
t.Fatal(err)
}
defer rows.Close()
for id := 1; rows.Next(); id++ {
var gotID, attempts int
var status string
var next sql.NullTime
if err := rows.Scan(&gotID, &status, &attempts, &next); err != nil {
t.Fatal(err)
}
if gotID == 1 {
if status != "pending" || attempts != 0 || next.Valid {
t.Fatalf("matching event = %q, %d, %v", status, attempts, next)
}
} else if status != "failed" || attempts != 10 || !next.Valid {
t.Fatalf("event %d was unexpectedly changed: %q, %d, %v", gotID, status, attempts, next)
}
}
}
func TestRollbackMigrationRefusesRenameOperations(t *testing.T) {
db, err := sql.Open("sqlite", filepath.Join(t.TempDir(), "rollback.db"))
if err != nil {
@@ -511,7 +560,7 @@ func TestOnlineBackupCanBeOpenedReadOnly(t *testing.T) {
t.Fatalf("backup mode = %v", info.Mode().Perm())
}
version, err := InspectDatabase(ctx, backup)
if err != nil || version != 2 {
if err != nil || version != 3 {
t.Fatalf("backup version = %d, %v", version, err)
}
}