Merged in feature/localstackcompose (pull request #49)

Add Localstack to Compose * localstack * addedlocalstackpluscleanup#
2025-02-05 18:33:06 +00:00
parent 5c253b3592
commit 78dc2f6cbc
6 changed files with 121 additions and 50 deletions
@@ -19,6 +19,29 @@ services:
      retries: 5
    networks:
      - server-network
+  localstack:
+    image: localstack/localstack:4.1.0
+    ports:
+      - 4566:4566
+    environment:
+      AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
+      AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
+      AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN}
+      AWS_REGION: ${AWS_REGION}
+      SERVICES: "s3,sqs"
+      SKIP_SSL_CERT_DOWNLOAD: "1"
+      LOCALSTACK_HOST: localstack
+      SQS_ENDPOINT_STRATEGY: "path"
+      EAGER_SERVICE_LOADING: "1"
+    expose:
+      - 4566
+    # healthcheck:
+    #   test: ["CMD-SHELL", "pg_isready -U ${DB_USER}"]
+    #   interval: 10s
+    #   timeout: 5s
+    #   retries: 5
+    networks:
+      - server-network
 volumes:
  db-data:
 networks:
@@ -16,7 +16,8 @@
    "gotools@0.25.0",
    "nodePackages.jsonlint@1.6.3",
    "oapi-codegen@2.4.1",
-    "vacuum-go@0.14.1"
+    "vacuum-go@0.14.1",
+    "awscli2@2.19.0"
  ],
  "shell": {
    "init_hook": [
@@ -34,7 +35,16 @@
    "DB_NOSSL":              "true",
    "AWS_ACCESS_KEY_ID":     "test",
    "AWS_SECRET_ACCESS_KEY": "test",
-    "AWS_REGION":            "us-east-1"
+    "AWS_SESSION_TOKEN": "",
+    "AWS_REGION":            "us-east-1",
+    "AWS_ENDPOINT_URL":      "http://localhost:4566",
+    "QNAME_DOCUMENT_CLEAN":  "document_clean",
+    "QNAME_DOCUMENT_INIT":   "document_init",
+    "QNAME_QUERY_RUNNER":    "query_runner",
+    "DOCUMENT_CLEAN_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_clean",
+    "DOCUMENT_INIT_URL": "http://localstack:4566/queue/us-east-1/000000000000/document_init",
+    "QUERYRUNNER_URL": "http://localstack:4566/queue/us-east-1/000000000000/query_runner",
+    "BUCKET_IN":             "documentin"
  },
  "env_from": ".env"
 }
@@ -1,6 +1,70 @@
 {
  "lockfile_version": "1",
  "packages": {
+    "awscli2@2.19.0": {
+      "last_modified": "2024-12-03T12:40:06Z",
+      "resolved": "github:NixOS/nixpkgs/566e53c2ad750c84f6d31f9ccb9d00f823165550#awscli2",
+      "source": "devbox-search",
+      "version": "2.19.0",
+      "systems": {
+        "aarch64-darwin": {
+          "outputs": [
+            {
+              "name": "out",
+              "path": "/nix/store/ywgz36mwv7q52m3lnsj8d1h7x70nyylc-awscli2-2.19.0",
+              "default": true
+            },
+            {
+              "name": "dist",
+              "path": "/nix/store/254shki4ihfaqw0h1v5xj9vqcc0y5vkz-awscli2-2.19.0-dist"
+            }
+          ],
+          "store_path": "/nix/store/ywgz36mwv7q52m3lnsj8d1h7x70nyylc-awscli2-2.19.0"
+        },
+        "aarch64-linux": {
+          "outputs": [
+            {
+              "name": "out",
+              "path": "/nix/store/irx8wcn8hqar9nnq1h9b92cssl7plj1g-awscli2-2.19.0",
+              "default": true
+            },
+            {
+              "name": "dist",
+              "path": "/nix/store/yqnb36a2b6bmlpsbxx28sb5857r6j9dz-awscli2-2.19.0-dist"
+            }
+          ],
+          "store_path": "/nix/store/irx8wcn8hqar9nnq1h9b92cssl7plj1g-awscli2-2.19.0"
+        },
+        "x86_64-darwin": {
+          "outputs": [
+            {
+              "name": "out",
+              "path": "/nix/store/6jgmxb4gc0zbhhdd2h566vz4v08gz19g-awscli2-2.19.0",
+              "default": true
+            },
+            {
+              "name": "dist",
+              "path": "/nix/store/3zxc4fc3jqbd2m19p37ydpnj8w3s3mpm-awscli2-2.19.0-dist"
+            }
+          ],
+          "store_path": "/nix/store/6jgmxb4gc0zbhhdd2h566vz4v08gz19g-awscli2-2.19.0"
+        },
+        "x86_64-linux": {
+          "outputs": [
+            {
+              "name": "out",
+              "path": "/nix/store/g2w36iqqxj51nlfpid40dq102ag6kq8y-awscli2-2.19.0",
+              "default": true
+            },
+            {
+              "name": "dist",
+              "path": "/nix/store/r5rkanfiin4i6cicbhl5jfswydq9kqgw-awscli2-2.19.0-dist"
+            }
+          ],
+          "store_path": "/nix/store/g2w36iqqxj51nlfpid40dq102ag6kq8y-awscli2-2.19.0"
+        }
+      }
+    },
    "go-migrate@4.18.1": {
      "last_modified": "2024-12-23T21:10:33Z",
      "resolved": "github:NixOS/nixpkgs/de1864217bfa9b5845f465e771e0ecb48b30e02d#go-migrate",
@@ -20,7 +20,6 @@ func (c *Server) Listen(ctx context.Context) {
 		}
 	}()

-	// TODO need to slow down the polling so we are not spinning the cpu or burning up API calls.
 	for {
 		select {
 		case <-ctx.Done():
@@ -35,31 +34,6 @@ func (c *Server) Listen(ctx context.Context) {

 }

-// Alternative without go func if no pool is needed.
-// Just for comment for the next function which needs to be made safe.
-//func (c *Server) pollMessage(ctx context.Context) error {
-//	cfg := &queue.Config{
-//		URL:    c.queueURL,
-//		Client: c.client,
-//	}
-//	result, err := queue.Receive(ctx, cfg, []string{"type"})
-//	if err != nil {
-//		return fmt.Errorf("message fetch fail: %v", err)
-//	}
-//
-//	for _, message := range result.Messages {
-//		if err := c.controller.Process(ctx, &message); err != nil {
-//			return fmt.Errorf("message process fail: %v", err)
-//		}
-//
-//		if err := queue.Delete(ctx, cfg, &message); err != nil {
-//			return fmt.Errorf("message delete fail: %v", err)
-//		}
-//	}
-//
-//	return nil
-//}
-
 func (c *Server) pollMessage(ctx context.Context) error {
 	result, err := c.cfg.ReceiveFromQueue(ctx, &queue.ReceiveParams{
 		QueueURL: c.cfg.GetQueueURL(),
@@ -69,24 +43,20 @@ func (c *Server) pollMessage(ctx context.Context) error {
 	}

 	for _, message := range result.Messages {
-		// TODO Need to remove this go func() !!
-		// Must not have uncontrolled go routine creation.  See commented example above if
-		// simple case.
-		// if process takes too long to complete then we need to use a thread pool.
-		go func() {
-			err := c.cfg.GetController().Process(ctx, &message)
-			if err != nil {
-				slog.Error("message process fail", "err", err)
-			}
+		slog.Info("processing message", "id", message.MessageId)

-			err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{
-				QueueURL:      c.cfg.GetQueueURL(),
-				ReceiptHandle: message.ReceiptHandle,
-			})
-			if err != nil {
-				slog.Error("message delete fail", "err", err)
-			}
-		}()
+		err := c.cfg.GetController().Process(ctx, &message)
+		if err != nil {
+			slog.Error("message process fail", "err", err)
+		}
+
+		err = c.cfg.DeleteFromQueue(ctx, &queue.DeleteParams{
+			QueueURL:      c.cfg.GetQueueURL(),
+			ReceiptHandle: message.ReceiptHandle,
+		})
+		if err != nil {
+			slog.Error("message delete fail", "err", err)
+		}
 	}

 	return nil
@@ -11,11 +11,15 @@ tasks:
    cmds:
      - docker compose -f {{.COMPOSE_FILE}} build
  up:
-    cmds:
-      - docker compose -f {{.COMPOSE_FILE}} up --no-recreate
-  up:bg:
    cmds:
      - docker compose -f {{.COMPOSE_FILE}} up --no-recreate -d
+      - task compose:init
+  init:
+    cmds:
+      - aws sqs create-queue --queue-name $QNAME_DOCUMENT_CLEAN
+      - aws sqs create-queue --queue-name $QNAME_DOCUMENT_INIT
+      - aws sqs create-queue --queue-name $QNAME_QUERY_RUNNER
+      - aws s3 mb s3://$BUCKET_IN
  down:
    cmds:
      - docker compose -f {{.COMPOSE_FILE}} down
@@ -13,7 +13,7 @@ tasks:
      - sqlc generate --file sqlc.yml
  lint:
    cmds:
-      - task compose:up:bg
+      - task compose:up
      - sleep 2
      - sqlc vet --file sqlc.yml
  mig:create:
@@ -24,6 +24,6 @@ tasks:
        migrate create -ext sql -dir {{.MIGRATIONS}} $name
  mig:
    cmds:
-      - task compose:up:bg
+      - task compose:up
      - sleep 2
      - migrate -path {{.MIGRATIONS}} -database {{.DB_URI}} up