Release v1.0.2

.github/workflows/delete-images-from-ghcr.yml

@@ -0,0 +1,44 @@
+name: Delete Container Images from github container registry
+
+on:
+  workflow_dispatch:
+    inputs:
+      image_tag_to_delete: 
+        type: string
+        description: "image tag to delete"
+        required: true
+        default: "v0.0.0"
+
+permissions:
+  contents: read
+  packages: write
+
+jobs:
+  docker-builds:
+    strategy:
+      matrix:
+        app:
+          - accounts
+          - auth
+          - comms
+          - console
+          - container-registry
+          - iam
+          - infra
+          - worker-audit-logging
+          - webhook
+          - websocket-server
+          - message-office
+          - tenant-agent
+
+    name: Delete image from ghcr.io
+    runs-on: ubuntu-latest
+    steps:
+      - name: Delete image
+        uses: bots-house/ghcr-delete-image-action@v1.1.0
+        with:
+          owner: ${{ github.repository_owner }}
+          name:  ${{ github.event.repository.name }}/${{ matrix.app }}
+
+          token: ${{ secrets.GITHUB_TOKEN }}
+          tag: ${{ github.event.inputs.image_tag_to_delete }}

.tools/nvim/dap/go.lua

@@ -171,4 +171,17 @@ dap.configurations.go = {
       vim.g.nxt.project_root_dir .. "/apps/tenant-agent" .. "/.secrets/env",
     },
   },
+
+  {
+    type = "go",
+    name = "Debug Websocket Server",
+    request = "launch",
+    program = vim.g.nxt.project_root_dir .. "/apps/websocket-server",
+    args = { "--dev"},
+    console = "externalTerminal",
+    -- externalTerminal = true,
+    envFile = {
+      vim.g.nxt.project_root_dir .. "/apps/websocket-server" .. "/.secrets/env",
+    },
+  },
 }

apps/websocket-server/Taskfile.yml

@@ -4,7 +4,7 @@ dotenv:
   - .secrets/env
 
 vars:
-  ImagePrefix: "ghcr.io/kloudlite/platform/apis"
+  ImagePrefix: "ghcr.io/kloudlite/api"
 
 tasks:
   build:

apps/websocket-server/internal/app/app.go

@@ -38,7 +38,6 @@ var Module = fx.Module("app",
 			logr logging.Logger,
 			sessionRepo kv.Repo[*common.AuthSession],
 		) {
-
 			a := server.Raw()
 
 			a.Use(
@@ -76,7 +75,6 @@ var Module = fx.Module("app",
 			a.All("*", func(c *fiber.Ctx) error {
 				return c.SendStatus(fiber.StatusNotFound)
 			})
-
 		},
 	),
 )

apps/websocket-server/internal/domain/logs.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kloudlite/api/apps/websocket-server/internal/domain/types"
 	"github.com/kloudlite/api/apps/websocket-server/internal/domain/utils"
 	"github.com/kloudlite/api/pkg/errors"
+	"github.com/kloudlite/api/pkg/messaging/nats"
 	msg_nats "github.com/kloudlite/api/pkg/messaging/nats"
 	msg_types "github.com/kloudlite/api/pkg/messaging/types"
 
@@ -106,7 +107,6 @@ func (d *domain) handleLogsMsg(ctx types.Context, logsSubs *logs.LogsSubsMap, ms
 			}
 
 			go func() {
-
 				utils.WriteInfo(ctx, "subscribed to logs", msg.Id, types.ForLogs)
 
 				if err := jc.Consume(
@@ -148,7 +148,6 @@ func (d *domain) handleLogsMsg(ctx types.Context, logsSubs *logs.LogsSubsMap, ms
 				); err != nil {
 					utils.WriteError(ctx, err, msg.Id, types.ForLogs)
 				}
-
 			}()
 
 		}
@@ -163,6 +162,10 @@ func (d *domain) handleLogsMsg(ctx types.Context, logsSubs *logs.LogsSubsMap, ms
 						return err
 					}
 
+					if err := nats.DeleteConsumer(ctx.Context, d.jetStreamClient, res.Jc); err != nil {
+						return err
+					}
+
 					delete(*logsSubs, hash)
 				}
 				ctx.Mutex.Unlock()

apps/websocket-server/internal/domain/main.go

@@ -13,6 +13,7 @@ import (
 	"github.com/kloudlite/api/common"
 	"github.com/kloudlite/api/pkg/errors"
 	httpServer "github.com/kloudlite/api/pkg/http-server"
+	"github.com/kloudlite/api/pkg/messaging/nats"
 )
 
 func (d *domain) HandleWebSocket(ctx context.Context, c *websocket.Conn) error {
@@ -23,8 +24,8 @@ func (d *domain) HandleWebSocket(ctx context.Context, c *websocket.Conn) error {
 
 	mu := sync.Mutex{}
 
-	var logsSubs = &logs.LogsSubsMap{}
-	var rWatchSubs = &res_watch.ResWatchSubsMap{}
+	logsSubs := &logs.LogsSubsMap{}
+	rWatchSubs := &res_watch.ResWatchSubsMap{}
 
 	defer func() {
 		if err := c.Close(); err != nil {
@@ -35,7 +36,10 @@ func (d *domain) HandleWebSocket(ctx context.Context, c *websocket.Conn) error {
 			for _, v := range *logsSubs {
 				if v.Jc != nil {
 					if err := v.Jc.Stop(ctx); err != nil {
-						d.logger.Warnf("stop jetstream consumer: %w", err)
+						d.logger.Warnf("stop jetstream consumer failed with err: %w", err)
+					}
+					if err := nats.DeleteConsumer(ctx, d.jetStreamClient, v.Jc); err != nil {
+						d.logger.Warnf("deleting jetstream consumer failed with err: %w", err)
 					}
 				}
 			}
@@ -50,7 +54,6 @@ func (d *domain) HandleWebSocket(ctx context.Context, c *websocket.Conn) error {
 				}
 			}
 		}
-
 	}()
 
 	closed := false
@@ -67,7 +70,6 @@ func (d *domain) HandleWebSocket(ctx context.Context, c *websocket.Conn) error {
 	}
 
 	for {
-
 		if closed {
 			break
 		}

pkg/messaging/nats/jetstream-consumer.go

@@ -2,17 +2,19 @@ package nats
 
 import (
 	"context"
-	"github.com/kloudlite/api/pkg/errors"
 	"os"
 	"os/signal"
 
+	"github.com/kloudlite/api/pkg/errors"
+
 	"github.com/kloudlite/api/pkg/messaging/types"
 	"github.com/kloudlite/api/pkg/nats"
 	"github.com/nats-io/nats.go/jetstream"
 )
 
 type JetstreamConsumer struct {
 	name       string
+	stream     string
 	client     *nats.JetstreamClient
 	consumer   jetstream.Consumer
 	consumeCtx jetstream.ConsumeContext
@@ -114,5 +116,10 @@ func NewJetstreamConsumer(ctx context.Context, jc *nats.JetstreamClient, args Je
 		name:     args.ConsumerConfig.Name,
 		client:   jc,
 		consumer: c,
+		stream:   args.Stream,
 	}, nil
 }
+
+func DeleteConsumer(ctx context.Context, jc *nats.JetstreamClient, consumer *JetstreamConsumer) error {
+	return jc.Jetstream.DeleteConsumer(ctx, consumer.stream, consumer.name)
+}