Ilia Denisov
87 lines
2.8 KiB
Go
87 lines
2.8 KiB
Go
package shared
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"galaxy/authsession/internal/domain/devicesession"
|
|
"galaxy/authsession/internal/domain/gatewayprojection"
|
|
"galaxy/authsession/internal/ports"
|
|
"galaxy/authsession/internal/telemetry"
|
|
)
|
|
|
|
// PublishProjectionSnapshot publishes snapshot through publisher with a small
|
|
// bounded retry loop suitable for request-path consistency repair.
|
|
func PublishProjectionSnapshot(ctx context.Context, publisher ports.GatewaySessionProjectionPublisher, snapshot gatewayprojection.Snapshot) error {
|
|
return PublishProjectionSnapshotWithTelemetry(ctx, publisher, snapshot, nil, "")
|
|
}
|
|
|
|
// PublishProjectionSnapshotWithTelemetry publishes snapshot through publisher
|
|
// with the bounded request-path retry policy and optional publish-failure
|
|
// telemetry.
|
|
func PublishProjectionSnapshotWithTelemetry(
|
|
ctx context.Context,
|
|
publisher ports.GatewaySessionProjectionPublisher,
|
|
snapshot gatewayprojection.Snapshot,
|
|
telemetryRuntime *telemetry.Runtime,
|
|
operation string,
|
|
) error {
|
|
if publisher == nil {
|
|
return InternalError(errors.New("projection publisher must not be nil"))
|
|
}
|
|
if ctx == nil {
|
|
return ServiceUnavailable(errors.New("projection publish context must not be nil"))
|
|
}
|
|
if err := snapshot.Validate(); err != nil {
|
|
return InternalError(fmt.Errorf("publish projection snapshot: %w", err))
|
|
}
|
|
|
|
var lastErr error
|
|
for attempt := 0; attempt < MaxProjectionPublishAttempts; attempt++ {
|
|
if err := ctx.Err(); err != nil {
|
|
return ServiceUnavailable(err)
|
|
}
|
|
|
|
if err := publisher.PublishSession(ctx, snapshot); err == nil {
|
|
return nil
|
|
} else {
|
|
lastErr = err
|
|
}
|
|
}
|
|
|
|
telemetryRuntime.RecordProjectionPublishFailure(ctx, operation)
|
|
return ServiceUnavailable(
|
|
fmt.Errorf(
|
|
"publish projection snapshot %q after %d attempts: %w",
|
|
snapshot.DeviceSessionID,
|
|
MaxProjectionPublishAttempts,
|
|
lastErr,
|
|
),
|
|
)
|
|
}
|
|
|
|
// PublishSessionProjection converts record into the gateway-facing snapshot and
|
|
// publishes it with the bounded request-path retry policy.
|
|
func PublishSessionProjection(ctx context.Context, publisher ports.GatewaySessionProjectionPublisher, record devicesession.Session) error {
|
|
return PublishSessionProjectionWithTelemetry(ctx, publisher, record, nil, "")
|
|
}
|
|
|
|
// PublishSessionProjectionWithTelemetry converts record into the
|
|
// gateway-facing snapshot and publishes it with the bounded request-path retry
|
|
// policy and optional publish-failure telemetry.
|
|
func PublishSessionProjectionWithTelemetry(
|
|
ctx context.Context,
|
|
publisher ports.GatewaySessionProjectionPublisher,
|
|
record devicesession.Session,
|
|
telemetryRuntime *telemetry.Runtime,
|
|
operation string,
|
|
) error {
|
|
snapshot, err := ToGatewayProjectionSnapshot(record)
|
|
if err != nil {
|
|
return InternalError(err)
|
|
}
|
|
|
|
return PublishProjectionSnapshotWithTelemetry(ctx, publisher, snapshot, telemetryRuntime, operation)
|
|
}
|