story/internal/device/telemetry_integration_test.go

package device

import (
	"context"
	"encoding/json"
	"errors"
	"os"
	"testing"
	"time"

	"data-terminal-cloud/internal/database"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/idempotency"

	"github.com/jackc/pgx/v5/pgxpool"
)

const (
	telemetryDeviceID       = "94000000-0000-0000-0000-000000000501"
	telemetryDevicePublicID = "94000000-0000-0000-0000-000000000511"
	telemetryCommandID      = "94000000-0000-0000-0000-000000000601"
	telemetryCommandPublic  = "94000000-0000-0000-0000-000000000611"
)

func TestTelemetryServiceAtomicallyUpdatesTwinAndCommand(t *testing.T) {
	databaseURL := os.Getenv("TEST_DATABASE_URL")
	if databaseURL == "" {
		t.Skip("TEST_DATABASE_URL is not set")
	}
	ctx := context.Background()
	pool, err := database.Open(ctx, databaseURL, "device-telemetry-test")
	if err != nil {
		t.Fatal(err)
	}
	defer pool.Close()
	cleanupTelemetryFixture(t, ctx, pool)
	defer cleanupTelemetryFixture(t, context.Background(), pool)
	if _, err := pool.Exec(ctx, `INSERT INTO device(id,public_id,tenant_id,serial_digest,model,status,credential_ref)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',repeat('7',64),'telemetry-test','registered','test://telemetry')`, telemetryDeviceID, telemetryDevicePublicID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO device_twin(device_id,tenant_id,desired_revision,reported_revision)
		VALUES($1::uuid,'10000000-0000-0000-0000-000000000001',2,1)`, telemetryDeviceID); err != nil {
		t.Fatal(err)
	}
	if _, err := pool.Exec(ctx, `INSERT INTO device_command(id,public_id,tenant_id,device_id,command_type,payload,status,requested_by,idempotency_key,request_digest,expires_at,delivery_attempts)
		VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000001',$3::uuid,'diagnostics','{}','DELIVERED','10000000-0000-0000-0000-000000000101','telemetry-fixture',repeat('8',64),now()+interval '1 hour',1)`, telemetryCommandID, telemetryCommandPublic, telemetryDeviceID); err != nil {
		t.Fatal(err)
	}

	service, err := NewTelemetryService(pool, idempotency.NewExecutor(pool))
	if err != nil {
		t.Fatal(err)
	}
	identity := deviceauth.Identity{ID: telemetryDeviceID, PublicID: telemetryDevicePublicID, TenantID: "10000000-0000-0000-0000-000000000001", Status: "registered"}
	rssi := -51
	input := TelemetryBatchInput{Events: []TelemetryEvent{
		{EventKey: "status-1", SequenceNo: 1, OccurredAt: time.Date(2026, 8, 23, 12, 0, 0, 0, time.UTC), EventType: "status", Status: &TelemetryStatus{
			FirmwareVersion: "1.2.3", ReportedConfigRevision: 1, Network: TelemetryNetwork{Type: "wifi", RSSI: &rssi},
			RTC: TelemetryRTC{}, RecordingQueue: TelemetryRecordingQueue{PendingSegments: 2},
		}},
		{EventKey: "error-2", SequenceNo: 2, OccurredAt: time.Date(2026, 8, 23, 12, 0, 1, 0, time.UTC), EventType: "error", Error: &TelemetryError{Code: "RTC_RECONNECTING", Severity: "warning"}},
		{EventKey: "command-3", SequenceNo: 3, OccurredAt: time.Date(2026, 8, 23, 12, 0, 2, 0, time.UTC), EventType: "command_result", CommandResult: &TelemetryCommandResult{
			CommandID: telemetryCommandPublic, Status: "SUCCEEDED", ResultSummary: map[string]any{"result_code": "OK"},
		}},
	}}
	body, _ := json.Marshal(input)
	response, replayed, err := service.Submit(ctx, identity, "telemetry-idempotency-0001", body, input)
	if err != nil || replayed || response.AcceptedCount != 3 || response.DuplicateCount != 0 || len(response.Actions) != 1 || response.Actions[0].Type != "refresh_config" {
		t.Fatalf("response=%#v replayed=%v err=%v", response, replayed, err)
	}
	replayedResponse, replayed, err := service.Submit(ctx, identity, "telemetry-idempotency-0001", body, input)
	if err != nil || !replayed || replayedResponse.AcceptedCount != 3 {
		t.Fatalf("replayed response=%#v replayed=%v err=%v", replayedResponse, replayed, err)
	}
	duplicateResponse, replayed, err := service.Submit(ctx, identity, "telemetry-idempotency-0002", body, input)
	if err != nil || replayed || duplicateResponse.AcceptedCount != 0 || duplicateResponse.DuplicateCount != 3 {
		t.Fatalf("duplicate response=%#v replayed=%v err=%v", duplicateResponse, replayed, err)
	}
	conflicting := input
	conflicting.Events = append([]TelemetryEvent(nil), input.Events...)
	conflicting.Events[1].Error = &TelemetryError{Code: "RTC_FAILED", Severity: "error"}
	conflictingBody, _ := json.Marshal(conflicting)
	if _, _, err := service.Submit(ctx, identity, "telemetry-idempotency-0003", conflictingBody, conflicting); err == nil {
		t.Fatal("conflicting telemetry was accepted")
	} else {
		var conflict *TelemetryConflictError
		if !errors.As(err, &conflict) || len(conflict.Conflicts) != 1 {
			t.Fatalf("conflicting telemetry error=%v", err)
		}
	}
	ahead := TelemetryBatchInput{Events: []TelemetryEvent{{EventKey: "status-4", SequenceNo: 4, OccurredAt: time.Now().UTC(), EventType: "status", Status: &TelemetryStatus{
		FirmwareVersion: "1.2.3", ReportedConfigRevision: 3, Network: TelemetryNetwork{Type: "wifi"}, RTC: TelemetryRTC{}, RecordingQueue: TelemetryRecordingQueue{},
	}}}}
	aheadBody, _ := json.Marshal(ahead)
	if _, _, err := service.Submit(ctx, identity, "telemetry-idempotency-0004", aheadBody, ahead); !errors.Is(err, ErrTelemetryConfigAhead) {
		t.Fatalf("ahead configuration error=%v", err)
	}

	var telemetrySequence int64
	var firmware, errorCode, errorSeverity, commandStatus string
	var events int
	if err := pool.QueryRow(ctx, `SELECT telemetry_sequence,firmware_version,last_error_code,last_error_severity FROM device_twin WHERE device_id=$1::uuid`, telemetryDeviceID).Scan(&telemetrySequence, &firmware, &errorCode, &errorSeverity); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT status FROM device_command WHERE id=$1::uuid`, telemetryCommandID).Scan(&commandStatus); err != nil {
		t.Fatal(err)
	}
	if err := pool.QueryRow(ctx, `SELECT count(*) FROM device_telemetry_event WHERE device_id=$1::uuid`, telemetryDeviceID).Scan(&events); err != nil {
		t.Fatal(err)
	}
	if telemetrySequence != 3 || firmware != "1.2.3" || errorCode != "RTC_RECONNECTING" || errorSeverity != "warning" || commandStatus != "SUCCEEDED" || events != 3 {
		t.Fatalf("sequence=%d firmware=%s error=%s/%s command=%s events=%d", telemetrySequence, firmware, errorCode, errorSeverity, commandStatus, events)
	}
}

func cleanupTelemetryFixture(t *testing.T, ctx context.Context, pool *pgxpool.Pool) {
	t.Helper()
	statements := []struct {
		query string
		arg   string
	}{
		{`DELETE FROM device_telemetry_event WHERE device_id=$1::uuid`, telemetryDeviceID},
		{`DELETE FROM idempotency_record WHERE actor_key=$1`, "device:" + telemetryDevicePublicID},
		{`DELETE FROM device_presence_state WHERE device_id=$1::uuid`, telemetryDeviceID},
		{`DELETE FROM device_command WHERE device_id=$1::uuid`, telemetryDeviceID},
		{`DELETE FROM device_twin WHERE device_id=$1::uuid`, telemetryDeviceID},
		{`DELETE FROM device WHERE id=$1::uuid`, telemetryDeviceID},
	}
	for _, statement := range statements {
		if _, err := pool.Exec(ctx, statement.query, statement.arg); err != nil {
			t.Fatalf("cleanup telemetry fixture: %v", err)
		}
	}
}