Parity sweep 3

.github/workflows/ci.yml

@@ -74,6 +74,24 @@ jobs:
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v9
 
+  modernize:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          token: ${{ env.GH_CI_TOKEN }}
+          ref: ${{ github.event.workflow_run.head_sha || github.sha }}
+
+      - name: Set up Go
+        uses: actions/setup-go@v6
+        with:
+          go-version-file: go.mod
+          check-latest: true
+          cache: true
+
+      - name: go fix (modernize)
+        run: go fix -diff ./...
+
   govulncheck:
     runs-on: ubuntu-latest
     steps:

.gitignore

@@ -55,3 +55,9 @@ lambda.zip
 .terraform.lock.hcl
 .terraform.lock.hcl
 /releaser
+
+# personal local Claude Code settings (never sync bypass perms to origin)
+.claude/settings.local.json
+
+# session resume checkpoint (transient)
+checkpoint.md

cli.go

@@ -2584,6 +2584,10 @@ func initializeServices(appCtx *service.AppContext) ([]service.Registerable, err
 	// Wire SNS→SQS delivery: when SNS publishes a message, deliver it to SQS queues.
 	wireSNSToSQS(byName["SNS"], byName["SQS"])
 
+	// Wire SNS→Lambda and SNS→Firehose subscription delivery, plus the SQS sender
+	// used for Lambda/Firehose subscription DLQ redelivery.
+	wireSNSToLambdaFirehose(byName["SNS"], byName["Lambda"], byName["Firehose"], byName["SQS"])
+
 	// Wire SQS → CloudWatch metric emission for NumberOfMessagesSent/Received/Deleted.
 	wireSQSMetrics(byName["SQS"], byName["CloudWatch"])
 
@@ -3145,6 +3149,54 @@ func wireSNSToSQS(snsReg, sqsReg service.Registerable) {
 	sqsBk.SubscribeToSNS(emitter)
 }
 
+// wireSNSToLambdaFirehose connects the SNS backend to Lambda and Firehose so that
+// Lambda- and Firehose-protocol subscriptions actually receive published messages,
+// and wires an SQS sender so failed Lambda/Firehose deliveries with a RedrivePolicy
+// land in the subscription's dead-letter queue. This is independent of wireSNSToSQS:
+// that function wires the SNS→SQS *subscription* delivery path via a publish emitter,
+// while the SQS sender wired here only serves DLQ redelivery on failed Lambda/Firehose
+// invocations (SNS backend's sqsSender field), so there is no overlap or double-wiring.
+func wireSNSToLambdaFirehose(snsReg, lambdaReg, firehoseReg, sqsReg service.Registerable) {
+	snsH, ok := snsReg.(*snsbackend.Handler)
+	if !ok {
+		return
+	}
+
+	snsBk, ok := snsH.Backend.(*snsbackend.InMemoryBackend)
+	if !ok {
+		return
+	}
+
+	if lambdaH, lambdaOk := lambdaReg.(*lambdabackend.Handler); lambdaOk {
+		if lambdaBk, bkOk := lambdaH.Backend.(*lambdabackend.InMemoryBackend); bkOk {
+			snsBk.SetLambdaBackend(lambdaBk)
+		}
+	}
+
+	if firehoseH, firehoseOk := firehoseReg.(*firehosebackend.Handler); firehoseOk {
+		if firehoseBk, bkOk := firehoseH.Backend.(*firehosebackend.InMemoryBackend); bkOk {
+			snsBk.SetFirehoseBackend(&snsFirehosePutterAdapter{backend: firehoseBk})
+		}
+	}
+
+	if sqsH, sqsOk := sqsReg.(*sqsbackend.Handler); sqsOk {
+		if sqsBk, bkOk := sqsH.Backend.(*sqsbackend.InMemoryBackend); bkOk {
+			snsBk.SetSQSSender(&sqsSenderAdapter{backend: sqsBk})
+		}
+	}
+}
+
+// snsFirehosePutterAdapter adapts the Firehose backend to the sns.FirehosePutter
+// interface, which omits the context parameter that the Firehose backend's
+// PutRecordBatch requires.
+type snsFirehosePutterAdapter struct {
+	backend *firehosebackend.InMemoryBackend
+}
+
+func (a *snsFirehosePutterAdapter) PutRecordBatch(streamName string, records [][]byte) (int, error) {
+	return a.backend.PutRecordBatch(context.Background(), streamName, records)
+}
+
 // wireSQSMetrics wires the CloudWatch metric emitter into the SQS backend so that
 // SendMessage, ReceiveMessage, and DeleteMessage operations emit CloudWatch metrics.
 func wireSQSMetrics(sqsReg, cwReg service.Registerable) {
@@ -3165,7 +3217,7 @@ func wireSQSMetrics(sqsReg, cwReg service.Registerable) {
 	sqsBk.SetMetricEmitter(
 		sqsbackend.MetricEmitterFunc(
 			func(namespace, name string, value float64, unit string) error {
-				_, err := cwBk.PutMetricData(namespace, []cwbackend.MetricDatum{
+				err := cwBk.PutMetricData(namespace, []cwbackend.MetricDatum{
 					{
 						MetricName: name,
 						Value:      value,
@@ -3309,7 +3361,10 @@ func (a *s3EventBridgeAdapter) PublishS3Event(
 	ctx context.Context,
 	source, detailType, detail string,
 ) {
-	a.backend.PutEvents(ctx, []ebbackend.EventEntry{
+	// Best-effort S3 -> EventBridge forwarding: PublishS3Event has no error
+	// channel (s3.EventBridgePublisher is fire-and-forget), so the PutEvents
+	// error (added by the Phase 3.3 store conversion) is explicitly discarded.
+	_, _ = a.backend.PutEvents(ctx, []ebbackend.EventEntry{
 		{Source: source, DetailType: detailType, Detail: detail},
 	})
 }
@@ -4204,7 +4259,7 @@ func wireCWLogsMetricEmitter(cwlogsReg, cwReg service.Registerable) {
 	cwlogsBk.SetMetricEmitter(
 		cwlogsbackend.MetricEmitterFunc(
 			func(namespace, name string, value float64, unit string) error {
-				_, err := cwBk.PutMetricData(namespace, []cwbackend.MetricDatum{
+				err := cwBk.PutMetricData(namespace, []cwbackend.MetricDatum{
 					{
 						MetricName: name,
 						Namespace:  namespace,

cli_adapters.go

@@ -40,9 +40,9 @@ func (a *schedEventBusAdapter) PutSchedulerEvent(
 			Time:         &now,
 		},
 	}
-	a.backend.PutEvents(ctx, entries)
+	_, err := a.backend.PutEvents(ctx, entries)
 
-	return nil
+	return err
 }
 
 type schedKinesisAdapter struct {

cli_sns_wiring_test.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	s3sdk "github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/blackbirdworks/gopherstack/pkgs/arn"
+	"github.com/blackbirdworks/gopherstack/pkgs/config"
+	firehosebackend "github.com/blackbirdworks/gopherstack/services/firehose"
+	lambdabackend "github.com/blackbirdworks/gopherstack/services/lambda"
+	snsbackend "github.com/blackbirdworks/gopherstack/services/sns"
+	sqsbackend "github.com/blackbirdworks/gopherstack/services/sqs"
+)
+
+// wiringMockS3Storer implements firehose.S3Storer so the test can observe that a
+// record delivered through the SNS→Firehose path actually reached S3.
+type wiringMockS3Storer struct {
+	calls []wiringS3PutCall
+}
+
+type wiringS3PutCall struct {
+	bucket string
+	key    string
+	body   []byte
+}
+
+func (m *wiringMockS3Storer) PutObject(
+	_ context.Context,
+	input *s3sdk.PutObjectInput,
+) (*s3sdk.PutObjectOutput, error) {
+	body, _ := io.ReadAll(input.Body)
+	m.calls = append(m.calls, wiringS3PutCall{
+		bucket: aws.ToString(input.Bucket),
+		key:    aws.ToString(input.Key),
+		body:   body,
+	})
+
+	return &s3sdk.PutObjectOutput{}, nil
+}
+
+// TestWireSNSToLambdaFirehose_EndToEndDelivery exercises the exact composition-root
+// wiring used by cli.go's setup (wireSNSToLambdaFirehose) against REAL Lambda, Firehose,
+// and SQS backends — not test doubles — to prove that subscribing a Lambda function or
+// Firehose delivery stream to an SNS topic and publishing actually delivers, instead of
+// silently no-opping as it did before SetLambdaBackend/SetFirehoseBackend/SetSQSSender
+// were wired in the running binary.
+func TestWireSNSToLambdaFirehose_EndToEndDelivery(t *testing.T) {
+	t.Parallel()
+
+	snsBk := snsbackend.NewInMemoryBackend()
+	snsH := snsbackend.NewHandler(snsBk)
+
+	lambdaBk := lambdabackend.NewInMemoryBackend(
+		nil, nil, lambdabackend.DefaultSettings(), config.DefaultAccountID, config.DefaultRegion,
+	)
+	lambdaH := lambdabackend.NewHandler(lambdaBk)
+
+	firehoseBk := firehosebackend.NewInMemoryBackend(config.DefaultAccountID, config.DefaultRegion)
+	firehoseH := firehosebackend.NewHandler(firehoseBk)
+
+	sqsBk := sqsbackend.NewInMemoryBackend()
+	sqsH := sqsbackend.NewHandler(sqsBk)
+
+	// This is the exact call cli.go's setupServices makes.
+	wireSNSToLambdaFirehose(snsH, lambdaH, firehoseH, sqsH)
+
+	ctx := context.Background()
+
+	// --- SNS -> Firehose: publish must reach the real Firehose backend and flush to S3. ---
+	s3mock := &wiringMockS3Storer{}
+	firehoseBk.SetS3Backend(s3mock)
+
+	_, err := firehoseBk.CreateDeliveryStream(ctx, firehosebackend.CreateDeliveryStreamInput{
+		Name: "wiring-test-stream",
+		S3Destination: &firehosebackend.S3DestinationDescription{
+			BucketARN: "arn:aws:s3:::wiring-test-bucket",
+			BufferingHints: &firehosebackend.BufferingHints{
+				SizeInMBs:         128,
+				IntervalInSeconds: 900,
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	firehoseTopic, err := snsBk.CreateTopic("firehose-topic", nil)
+	require.NoError(t, err)
+
+	firehoseStreamARN := arn.Build(
+		"firehose", config.DefaultRegion, config.DefaultAccountID, "deliverystream/wiring-test-stream",
+	)
+	_, err = snsBk.Subscribe(firehoseTopic.TopicArn, "firehose", firehoseStreamARN, "")
+	require.NoError(t, err)
+
+	_, err = snsBk.Publish(firehoseTopic.TopicArn, "hello-firehose", "", "", nil)
+	require.NoError(t, err)
+
+	firehoseBk.FlushAll(ctx)
+
+	require.Len(t, s3mock.calls, 1,
+		"SNS publish must have been delivered to the real Firehose backend via cli.go's wireSNSToLambdaFirehose")
+	assert.Contains(t, string(s3mock.calls[0].body), "hello-firehose")
+
+	// --- SNS -> Lambda with DLQ: publish must reach the real Lambda backend. There is
+	// no Docker runtime available in this test, so the invocation fails deterministically
+	// (a real failure mode, not a stub) — proving delivery requires the failed message to
+	// be redirected to the real SQS DLQ via the wired SQS sender. ---
+	dlqOut, err := sqsBk.CreateQueue(&sqsbackend.CreateQueueInput{QueueName: "lambda-dlq"})
+	require.NoError(t, err)
+
+	lambdaTopic, err := snsBk.CreateTopic("lambda-topic", nil)
+	require.NoError(t, err)
+
+	functionARN := arn.Build(
+		"lambda", config.DefaultRegion, config.DefaultAccountID, "function:wiring-test-fn",
+	)
+	sub, err := snsBk.Subscribe(lambdaTopic.TopicArn, "lambda", functionARN, "")
+	require.NoError(t, err)
+
+	dlqARN := arn.Build("sqs", config.DefaultRegion, config.DefaultAccountID, "lambda-dlq")
+	redrivePolicy, err := json.Marshal(map[string]string{"deadLetterTargetArn": dlqARN})
+	require.NoError(t, err)
+	require.NoError(t, snsBk.SetSubscriptionAttributes(sub.SubscriptionArn, "RedrivePolicy", string(redrivePolicy)))
+
+	_, err = snsBk.Publish(lambdaTopic.TopicArn, "hello-lambda", "", "", nil)
+	require.NoError(t, err)
+
+	recvOut, err := sqsBk.ReceiveMessage(&sqsbackend.ReceiveMessageInput{
+		QueueURL:            dlqOut.QueueURL,
+		MaxNumberOfMessages: 1,
+	})
+	require.NoError(t, err)
+	require.Len(t, recvOut.Messages, 1,
+		"failed Lambda delivery must have been redirected to the real SQS DLQ via cli.go's wireSNSToLambdaFirehose "+
+			"(proves both SetLambdaBackend and SetSQSSender were wired)")
+	assert.Equal(t, "hello-lambda", recvOut.Messages[0].Body)
+}