Method WrapStreamingHandler

backend/api/auth/auth.go:115–150

WrapStreamingHandler implements the ConnectRPC interceptor interface for streaming handlers.

(next connect.StreamingHandlerFunc)

// WrapStreamingHandler implements the ConnectRPC interceptor interface for streaming handlers.
func (in *APIAuthInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc {
	return func(ctx context.Context, conn connect.StreamingHandlerConn) error {
		accessTokenStr, err := GetTokenFromHeaders(conn.RequestHeader())
		if err != nil {
			return connect.NewError(connect.CodeUnauthenticated, err)
		}

		authContext, err := getAuthContext(conn.Spec().Procedure)
		if err != nil {
			return err
		}
		ctx = context.WithValue(ctx, common.AuthContextKey, authContext)

		user, claims, err := in.authenticate(ctx, accessTokenStr)
		if err != nil {
			if IsAuthenticationSkipped(conn.Spec().Procedure, authContext) {
				return next(ctx, conn)
			}
			return connect.NewError(connect.CodeUnauthenticated, err)
		}

		in.profile.LastActiveTS.Store(time.Now().Unix())
		ctx = context.WithValue(ctx, common.UserContextKey, user)
		ctx = context.WithValue(ctx, common.WorkspaceIDContextKey, claims.WorkspaceID)

		var tokenExpiry time.Time
		if claims.ExpiresAt != nil {
			tokenExpiry = claims.ExpiresAt.Time
		}

		return next(ctx, &authStreamingConn{
			StreamingHandlerConn: conn,
			tokenExpiry:          tokenExpiry,
		})
	}
}

// authStreamingConn wraps a streaming connection to check token expiry on every received message.
type authStreamingConn struct {