fix(tokenizer): persist last_observed_at on postgres

Values in the flush CTE were untyped literals, which postgres resolves
to text and refuses to assign to a timestamptz column. Building the
CTE from the token model lets bun cast per dialect.

pkg/tokenizer/opaquetokenizer/provider.go

@@ -364,12 +364,26 @@ func (provider *provider) gc(ctx context.Context, org *types.Organization) error
 }
 
 func (provider *provider) flushLastObservedAt(ctx context.Context, org *types.Organization) error {
-	accessTokenToLastObservedAt, err := provider.listLastObservedAtDesc(ctx, org.ID)
+	tokens, err := provider.tokenStore.ListByOrgID(ctx, org.ID)
 	if err != nil {
 		return err
 	}
 
-	if err := provider.tokenStore.UpdateLastObservedAtByAccessToken(ctx, accessTokenToLastObservedAt); err != nil {
+	observedTokens := make([]*authtypes.StorableToken, 0, len(tokens))
+	for _, token := range tokens {
+		cachedLastObservedAt, ok := provider.lastObservedAtCache.Get(lastObservedAtCacheKey(token.AccessToken, token.UserID))
+		if !ok {
+			continue
+		}
+
+		if err := token.UpdateLastObservedAt(cachedLastObservedAt); err != nil {
+			continue
+		}
+
+		observedTokens = append(observedTokens, token)
+	}
+
+	if err := provider.tokenStore.UpdateLastObservedAt(ctx, observedTokens); err != nil {
 		return err
 	}
 

pkg/tokenizer/tokenizerstore/sqltokenizerstore/store.go

@@ -232,15 +232,16 @@ func (store *store) ListByUserID(ctx context.Context, userID valuer.UUID) ([]*au
 	return tokens, nil
 }
 
-func (store *store) UpdateLastObservedAtByAccessToken(ctx context.Context, accessTokenToLastObservedAt []map[string]any) error {
-	if len(accessTokenToLastObservedAt) == 0 {
+func (store *store) UpdateLastObservedAt(ctx context.Context, tokens []*authtypes.StorableToken) error {
+	if len(tokens) == 0 {
 		return nil
 	}
 
 	values := store.
 		sqlstore.
 		BunDBCtx(ctx).
-		NewValues(&accessTokenToLastObservedAt)
+		NewValues(&tokens).
+		Column("id", "last_observed_at", "updated_at")
 
 	_, err := store.
 		sqlstore.
@@ -250,8 +251,8 @@ func (store *store) UpdateLastObservedAtByAccessToken(ctx context.Context, acces
 		Model((*authtypes.StorableToken)(nil)).
 		TableExpr("update_cte").
 		Set("last_observed_at = update_cte.last_observed_at").
-		Where("auth_token.access_token = update_cte.access_token").
-		Where("auth_token.user_id = update_cte.user_id").
+		Set("updated_at = update_cte.updated_at").
+		Where("auth_token.id = update_cte.id").
 		Exec(ctx)
 	if err != nil {
 		return err

pkg/tokenizer/tokenizerstore/sqltokenizerstore/store_test.go

@@ -0,0 +1,74 @@
+package sqltokenizerstore
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/SigNoz/signoz/pkg/sqlstore"
+	"github.com/SigNoz/signoz/pkg/sqlstore/sqlstoretest"
+	"github.com/SigNoz/signoz/pkg/types/authtypes"
+	"github.com/SigNoz/signoz/pkg/valuer"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUpdateLastObservedAt(t *testing.T) {
+	testCases := []struct {
+		name          string
+		provider      string
+		tokens        []*authtypes.StorableToken
+		expectedQuery string
+	}{
+		{
+			name:          "Sqlite_Empty",
+			provider:      "sqlite",
+			tokens:        nil,
+			expectedQuery: "",
+		},
+		{
+			name:          "Postgres_Empty",
+			provider:      "postgres",
+			tokens:        []*authtypes.StorableToken{},
+			expectedQuery: "",
+		},
+		{
+			name:     "Sqlite_OneToken",
+			provider: "sqlite",
+			tokens: []*authtypes.StorableToken{
+				{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000001"), AccessToken: "access-one", RefreshToken: "refresh-one", LastObservedAt: time.Date(2026, 9, 22, 10, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 10, 0, 1, 0, time.UTC)},
+			},
+			expectedQuery: `WITH "update_cte" ("id", "last_observed_at", "updated_at") AS (VALUES ('019984d1-0000-7000-8000-000000000001', '2026-09-22 10:00:00+00:00', '2026-09-22 10:00:01+00:00')) UPDATE "auth_token" AS "auth_token" SET last_observed_at = update_cte.last_observed_at, updated_at = update_cte.updated_at FROM update_cte WHERE (auth_token.id = update_cte.id)`,
+		},
+		{
+			name:     "Postgres_TwoTokens",
+			provider: "postgres",
+			tokens: []*authtypes.StorableToken{
+				{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000002"), AccessToken: "access-two", RefreshToken: "refresh-two", LastObservedAt: time.Date(2026, 9, 22, 11, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 11, 0, 1, 0, time.UTC)},
+				{ID: valuer.MustNewUUID("019984d1-0000-7000-8000-000000000003"), AccessToken: "access-three", RefreshToken: "refresh-three", LastObservedAt: time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC), UpdatedAt: time.Date(2026, 9, 22, 12, 0, 1, 0, time.UTC)},
+			},
+			expectedQuery: `WITH "update_cte" ("id", "last_observed_at", "updated_at") AS (VALUES ('019984d1-0000-7000-8000-000000000002'::text, '2026-09-22 11:00:00+00:00'::TIMESTAMPTZ, '2026-09-22 11:00:01+00:00'::TIMESTAMPTZ), ('019984d1-0000-7000-8000-000000000003'::text, '2026-09-22 12:00:00+00:00'::TIMESTAMPTZ, '2026-09-22 12:00:01+00:00'::TIMESTAMPTZ)) UPDATE "auth_token" AS "auth_token" SET last_observed_at = update_cte.last_observed_at, updated_at = update_cte.updated_at FROM update_cte WHERE (auth_token.id = update_cte.id)`,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			var executedQuery string
+			matcher := sqlmock.QueryMatcherFunc(func(_, actual string) error {
+				executedQuery = actual
+				return nil
+			})
+
+			sqlStore := sqlstoretest.New(sqlstore.Config{Provider: testCase.provider}, matcher)
+			if testCase.expectedQuery != "" {
+				sqlStore.Mock().ExpectExec("").WillReturnResult(sqlmock.NewResult(0, int64(len(testCase.tokens))))
+			}
+
+			err := NewStore(sqlStore).UpdateLastObservedAt(context.Background(), testCase.tokens)
+			require.NoError(t, err)
+			require.NoError(t, sqlStore.Mock().ExpectationsWereMet())
+			assert.Equal(t, testCase.expectedQuery, executedQuery)
+		})
+	}
+}

pkg/types/authtypes/token.go

@@ -258,6 +258,6 @@ type TokenStore interface {
 	// Delete a token by userID.
 	DeleteByUserID(context.Context, valuer.UUID) error
 
-	// Update last observed at by access token.
-	UpdateLastObservedAtByAccessToken(context.Context, []map[string]any) error
+	// Update last observed at of the given tokens.
+	UpdateLastObservedAt(context.Context, []*StorableToken) error
 }