diff --git a/.vscode/terminals.json b/.vscode/terminals.json index c2565fb8783..876773fca36 100644 --- a/.vscode/terminals.json +++ b/.vscode/terminals.json @@ -30,13 +30,6 @@ "cwd": "packages/web/app", "command": "pnpm dev" }, - { - "name": "tokens:dev", - "description": "Run tokens service", - "open": true, - "cwd": "packages/services/tokens", - "command": "pnpm dev" - }, { "name": "schema:dev", "description": "Run schema service", diff --git a/deployment/index.ts b/deployment/index.ts index c184dcdb0ac..a14ca44d1a8 100644 --- a/deployment/index.ts +++ b/deployment/index.ts @@ -24,7 +24,6 @@ import { deployS3, deployS3AuditLog, deployS3Mirror } from './services/s3'; import { deploySchema } from './services/schema'; import { configureSentry } from './services/sentry'; import { configureSlackApp } from './services/slack-app'; -import { deployTokens } from './services/tokens'; import { deployUsage } from './services/usage'; import { deployUsageIngestor } from './services/usage-ingestor'; import { deployWorkflows, PostmarkSecret } from './services/workflows'; @@ -123,18 +122,6 @@ const dbMigrations = deployDbMigrations({ dependencies: [databaseCleanupJob].filter(isDefined), }); -const tokens = deployTokens({ - image: docker.factory.getImageId('tokens', imagesTag), - environment, - dbMigrations, - docker, - postgres, - redis, - heartbeat: heartbeatsConfig.get('tokens'), - sentry, - observability, -}); - const commerce = deployCommerce({ image: docker.factory.getImageId('commerce', imagesTag), docker, @@ -150,7 +137,6 @@ const usage = deployUsage({ image: docker.factory.getImageId('usage', imagesTag), docker, environment, - tokens, redis, postgres, kafka, @@ -213,7 +199,6 @@ const graphql = deployGraphQL({ clickhouse, image: docker.factory.getImageId('server', imagesTag), docker, - tokens, schema, schemaPolicy, dbMigrations, @@ -340,7 +325,6 @@ deployCloudFlareSecurityTransform({ export const graphqlApiServiceId = graphql.service.id; export const usageApiServiceId = usage.service.id; export const usageIngestorApiServiceId = usageIngestor.service.id; -export const tokensApiServiceId = tokens.service.id; export const schemaApiServiceId = schema.service.id; export const appId = app.deployment.id; diff --git a/deployment/services/graphql.ts b/deployment/services/graphql.ts index d2abfed1209..1407fe09612 100644 --- a/deployment/services/graphql.ts +++ b/deployment/services/graphql.ts @@ -16,7 +16,6 @@ import { Redis } from './redis'; import { S3 } from './s3'; import { Schema } from './schema'; import { Sentry } from './sentry'; -import { Tokens } from './tokens'; import { Usage } from './usage'; import { Zendesk } from './zendesk'; @@ -31,7 +30,6 @@ export function deployGraphQL({ clickhouse, image, environment, - tokens, schema, schemaPolicy, cdn, @@ -55,7 +53,6 @@ export function deployGraphQL({ image: string; clickhouse: Clickhouse; environment: Environment; - tokens: Tokens; schema: Schema; schemaPolicy: SchemaPolicy; redis: Redis; @@ -132,7 +129,6 @@ export function deployGraphQL({ featureFlagsConfig.get('metricAlertRulesEnabled') ?? '0', REQUEST_LOGGING: '1', COMMERCE_ENDPOINT: serviceLocalEndpoint(commerce.service), - TOKENS_ENDPOINT: serviceLocalEndpoint(tokens.service), SCHEMA_ENDPOINT: serviceLocalEndpoint(schema.service), SCHEMA_POLICY_ENDPOINT: serviceLocalEndpoint(schemaPolicy.service), WEB_APP_URL: `https://${environment.appDns}`, diff --git a/deployment/services/tokens.ts b/deployment/services/tokens.ts deleted file mode 100644 index 990d6502853..00000000000 --- a/deployment/services/tokens.ts +++ /dev/null @@ -1,68 +0,0 @@ -import { ServiceDeployment } from '../utils/service-deployment'; -import { DbMigrations } from './db-migrations'; -import { Docker } from './docker'; -import { Environment } from './environment'; -import { Observability } from './observability'; -import { Postgres } from './postgres'; -import { Redis } from './redis'; -import { Sentry } from './sentry'; - -export type Tokens = ReturnType; - -export function deployTokens({ - environment, - dbMigrations, - heartbeat, - image, - docker, - postgres, - redis, - sentry, - observability, -}: { - observability: Observability; - image: string; - environment: Environment; - dbMigrations: DbMigrations; - heartbeat?: string; - docker: Docker; - redis: Redis; - postgres: Postgres; - sentry: Sentry; -}) { - return new ServiceDeployment( - 'tokens-service', - { - imagePullSecret: docker.secret, - readinessProbe: '/_readiness', - livenessProbe: '/_health', - startupProbe: '/_health', - exposesMetrics: true, - availabilityOnEveryNode: true, - replicas: environment.podsConfig.general.replicas, - image, - env: { - ...environment.envVars, - SENTRY: sentry.enabled ? '1' : '0', - HEARTBEAT_ENDPOINT: heartbeat ?? '', - OPENTELEMETRY_TRACE_USAGE_REQUESTS: observability.enabledForUsageService ? '1' : '', - OPENTELEMETRY_COLLECTOR_ENDPOINT: - observability.enabled && observability.tracingEndpoint - ? observability.tracingEndpoint - : '', - }, - }, - [dbMigrations], - ) - .withSecret('POSTGRES_HOST', postgres.pgBouncerSecret, 'host') - .withSecret('POSTGRES_PORT', postgres.pgBouncerSecret, 'port') - .withSecret('POSTGRES_USER', postgres.pgBouncerSecret, 'user') - .withSecret('POSTGRES_PASSWORD', postgres.pgBouncerSecret, 'password') - .withSecret('POSTGRES_DB', postgres.pgBouncerSecret, 'database') - .withSecret('POSTGRES_SSL', postgres.pgBouncerSecret, 'ssl') - .withSecret('REDIS_HOST', redis.secret, 'host') - .withSecret('REDIS_PORT', redis.secret, 'port') - .withSecret('REDIS_PASSWORD', redis.secret, 'password') - .withConditionalSecret(sentry.enabled, 'SENTRY_DSN', sentry.secret, 'dsn') - .deploy(); -} diff --git a/deployment/services/usage.ts b/deployment/services/usage.ts index f2a9dabf994..56721fd059c 100644 --- a/deployment/services/usage.ts +++ b/deployment/services/usage.ts @@ -9,13 +9,11 @@ import { Observability } from './observability'; import { Postgres } from './postgres'; import { Redis } from './redis'; import { Sentry } from './sentry'; -import { Tokens } from './tokens'; export type Usage = ReturnType; export function deployUsage({ environment, - tokens, postgres, redis, kafka, @@ -29,7 +27,6 @@ export function deployUsage({ observability: Observability; image: string; environment: Environment; - tokens: Tokens; postgres: Postgres; redis: Redis; kafka: Kafka; @@ -69,7 +66,6 @@ export function deployUsage({ KAFKA_BUFFER_INTERVAL: kafka.config.bufferInterval, KAFKA_BUFFER_DYNAMIC: kafkaBufferDynamic, KAFKA_TOPIC: kafka.config.topic, - TOKENS_ENDPOINT: serviceLocalEndpoint(tokens.service), COMMERCE_ENDPOINT: serviceLocalEndpoint(commerce.service), OPENTELEMETRY_COLLECTOR_ENDPOINT: observability.enabled && @@ -90,13 +86,7 @@ export function deployUsage({ maxReplicas: environment.podsConfig.usageService.maxReplicas, }, }, - [ - dbMigrations, - tokens.deployment, - tokens.service, - commerce.deployment, - commerce.service, - ].filter(Boolean), + [dbMigrations, commerce.deployment, commerce.service].filter(Boolean), ) // Redis .withSecret('REDIS_HOST', redis.secret, 'host') diff --git a/docker/docker-compose.community.yml b/docker/docker-compose.community.yml index 0eb5177dd27..fb4edd45b39 100644 --- a/docker/docker-compose.community.yml +++ b/docker/docker-compose.community.yml @@ -174,8 +174,6 @@ services: condition: service_completed_successfully s3_provision_buckets: condition: service_completed_successfully - tokens: - condition: service_healthy schema: condition: service_healthy policy: @@ -197,7 +195,6 @@ services: REDIS_HOST: redis REDIS_PORT: 6379 REDIS_PASSWORD: '${REDIS_PASSWORD}' - TOKENS_ENDPOINT: http://tokens:3003 SCHEMA_ENDPOINT: http://schema:3002 SCHEMA_POLICY_ENDPOINT: http://policy:3012 ENCRYPTION_SECRET: '${HIVE_ENCRYPTION_SECRET}' @@ -266,32 +263,6 @@ services: PROMETHEUS_METRICS: '${PROMETHEUS_METRICS:-}' LOG_LEVEL: '${LOG_LEVEL:-debug}' - tokens: - image: '${DOCKER_REGISTRY}tokens${DOCKER_TAG}' - networks: - - 'stack' - depends_on: - migrations: - condition: service_completed_successfully - environment: - NODE_ENV: production - POSTGRES_HOST: db - POSTGRES_USER: '${POSTGRES_USER}' - POSTGRES_PASSWORD: '${POSTGRES_PASSWORD}' - POSTGRES_PORT: 5432 - POSTGRES_DB: '${POSTGRES_DB}' - REDIS_HOST: redis - REDIS_PORT: 6379 - REDIS_PASSWORD: '${REDIS_PASSWORD}' - PORT: 3003 - SERVER_HOST: '${SERVER_HOST:-::}' - SERVER_HOST_IPV6_ONLY: '${SERVER_HOST_IPV6_ONLY:-0}' - LOG_LEVEL: '${LOG_LEVEL:-debug}' - OPENTELEMETRY_COLLECTOR_ENDPOINT: '${OPENTELEMETRY_COLLECTOR_ENDPOINT:-}' - SENTRY: '${SENTRY:-0}' - SENTRY_DSN: '${SENTRY_DSN:-}' - PROMETHEUS_METRICS: '${PROMETHEUS_METRICS:-}' - workflows: image: '${DOCKER_REGISTRY}workflows${DOCKER_TAG}' networks: @@ -328,13 +299,10 @@ services: depends_on: broker: condition: service_healthy - tokens: - condition: service_healthy ports: - 8081:3006 environment: NODE_ENV: production - TOKENS_ENDPOINT: http://tokens:3003 KAFKA_CONNECTION_MODE: 'docker' KAFKA_TOPIC: 'usage_reports_v2' KAFKA_BROKER: broker:29092 diff --git a/docker/docker.hcl b/docker/docker.hcl index 851512d8b07..d8d4d4f11f6 100644 --- a/docker/docker.hcl +++ b/docker/docker.hcl @@ -220,28 +220,6 @@ target "commerce" { ] } -target "tokens" { - inherits = ["service-base", get_target()] - contexts = { - dist = "${PWD}/packages/services/tokens/dist" - shared = "${PWD}/docker/shared" - } - args = { - SERVICE_DIR_NAME = "@hive/tokens" - IMAGE_TITLE = "graphql-hive/tokens" - IMAGE_DESCRIPTION = "The tokens service of the GraphQL Hive project." - PORT = "3003" - HEALTHCHECK_CMD = "wget --spider -q http://127.0.0.1:$${PORT}/_readiness" - } - tags = [ - local_image_tag("tokens"), - stable_image_tag("tokens"), - image_tag("tokens", COMMIT_SHA), - image_tag("tokens", COMMIT_SHORT_SHA), - image_tag("tokens", BRANCH_NAME) - ] -} - target "usage-ingestor" { inherits = ["service-base", get_target()] contexts = { @@ -391,7 +369,6 @@ group "build" { "schema", "policy", "storage", - "tokens", "usage-ingestor", "usage", "server", @@ -409,7 +386,6 @@ group "integration-tests" { "schema", "policy", "storage", - "tokens", "usage-ingestor", "usage", "server", diff --git a/integration-tests/docker-compose.integration.yaml b/integration-tests/docker-compose.integration.yaml index 412cf9739b8..056b6168196 100644 --- a/integration-tests/docker-compose.integration.yaml +++ b/integration-tests/docker-compose.integration.yaml @@ -223,8 +223,6 @@ services: depends_on: broker: condition: service_started # Redpand is ready when it starts - tokens: - condition: service_healthy usage-ingestor: environment: @@ -234,8 +232,6 @@ services: depends_on: broker: condition: service_started # Redpand is ready when it starts - tokens: - condition: service_healthy workflows: environment: diff --git a/integration-tests/tests/api/tokens.spec.ts b/integration-tests/tests/api/tokens.spec.ts index 873b31bc3c9..688323a1c08 100644 --- a/integration-tests/tests/api/tokens.spec.ts +++ b/integration-tests/tests/api/tokens.spec.ts @@ -1,7 +1,7 @@ +import { createHash, randomBytes } from 'node:crypto'; import { pollFor, readTokenInfo } from 'testkit/flow'; import { ProjectType } from 'testkit/gql/graphql'; -import { createTokenStorage } from '@hive/storage'; -import { generateToken } from '@hive/tokens'; +import { TargetTokenStorage } from '@hive/api/modules/token/providers/target-token-storage'; import { initSeed } from '../../testkit/seed'; test.concurrent('deleting a token should clear the cache', async () => { @@ -155,26 +155,26 @@ test.concurrent( const { createProject, organization } = await createOrg(); const { project, target } = await createProject(); - const tokenStorage = await createTokenStorage(seed.getPGConnectionString(), 1); - - try { - const token = generateToken(); - - // create new token so it does not yet exist in redis cache - const record = await tokenStorage.createToken({ - name: 'foo', - organization: organization.id, - project: project.id, - target: target.id, - scopes: [], - token: token.hash, - tokenAlias: token.alias, - }); - - // touch the token so it has a date - await tokenStorage.touchTokens({ tokens: [{ token: record.token, date: new Date() }] }); - const result = await readTokenInfo(token.secret).then(res => res.expectNoGraphQLErrors()); - expect(result.tokenInfo).toMatchInlineSnapshot(` + const token = randomBytes(16).toString('hex'); + const secret = createHash('sha256').update(token).digest('hex'); + const { pool } = await seed.createDbConnection(); + const tokenStorage = new TargetTokenStorage(pool); + + // create new token so it does not yet exist in redis cache + const record = await tokenStorage.createToken({ + name: 'foo', + organization: organization.id, + project: project.id, + target: target.id, + scopes: [], + token: secret, + tokenAlias: 'foobars', + }); + + // touch the token so it has a date + await TargetTokenStorage.touchTokenByHash({ pool })(secret); + const result = await readTokenInfo(token).then(res => res.expectNoGraphQLErrors()); + expect(result.tokenInfo).toMatchInlineSnapshot(` { __typename: TokenInfo, hasOrganizationDelete: false, @@ -197,8 +197,5 @@ test.concurrent( hasTargetTokensWrite: false, } `); - } finally { - await tokenStorage.destroy(); - } }, ); diff --git a/packages/services/api/package.json b/packages/services/api/package.json index f5ed2a41df4..d0306646ad4 100644 --- a/packages/services/api/package.json +++ b/packages/services/api/package.json @@ -11,7 +11,8 @@ "./modules/auth/providers/supertokens-store": "./src/modules/auth/providers/supertokens-store.ts", "./modules/shared/providers/logger": "./src/modules/shared/providers/logger.ts", "./modules/shared/providers/redis": "./src/modules/shared/providers/redis.ts", - "./modules/schema/providers/schema-version-store": "./src/modules/schema/providers/schema-version-store.ts" + "./modules/schema/providers/schema-version-store": "./src/modules/schema/providers/schema-version-store.ts", + "./modules/token/providers/target-token-storage": "./src/modules/token/providers/target-token-storage.ts" }, "peerDependencies": { "graphql": "^16.0.0", @@ -33,7 +34,6 @@ "@hive/schema": "workspace:*", "@hive/service-common": "workspace:*", "@hive/storage": "workspace:*", - "@hive/tokens": "workspace:*", "@hive/usage-common": "workspace:*", "@hive/usage-ingestor": "workspace:*", "@hive/workflows": "workspace:*", diff --git a/packages/services/api/src/create.ts b/packages/services/api/src/create.ts index 8546fa9c023..416af45d4aa 100644 --- a/packages/services/api/src/create.ts +++ b/packages/services/api/src/create.ts @@ -70,7 +70,6 @@ import { supportModule } from './modules/support'; import { provideSupportConfig, SupportConfig } from './modules/support/providers/config'; import { targetModule } from './modules/target'; import { tokenModule } from './modules/token'; -import { TOKENS_CONFIG, TokensConfig } from './modules/token/providers/tokens'; const modules = [ sharedModule, @@ -100,7 +99,6 @@ const modules = [ export function createRegistry({ app, commerce, - tokens, schemaService, schemaPolicyService, logger, @@ -129,7 +127,6 @@ export function createRegistry({ clickHouse: ClickHouseConfig; redis: Redis; commerce: CommerceConfig; - tokens: TokensConfig; schemaService: SchemaServiceConfig; schemaPolicyService: SchemaPolicyServiceConfig; githubApp: GitHubApplicationConfig | null; @@ -248,11 +245,6 @@ export function createRegistry({ useValue: clickHouse, scope: Scope.Singleton, }, - { - provide: TOKENS_CONFIG, - useValue: tokens, - scope: Scope.Singleton, - }, { provide: SCHEMA_SERVICE_CONFIG, useValue: schemaService, diff --git a/packages/services/api/src/modules/auth/lib/target-access-token-strategy.ts b/packages/services/api/src/modules/auth/lib/target-access-token-strategy.ts index 210947f99cd..88fed1e3e57 100644 --- a/packages/services/api/src/modules/auth/lib/target-access-token-strategy.ts +++ b/packages/services/api/src/modules/auth/lib/target-access-token-strategy.ts @@ -1,8 +1,8 @@ +import { PostgresDatabasePool } from '@hive/postgres'; import { maskToken, type FastifyReply, type FastifyRequest } from '@hive/service-common'; -import { AccessError } from '../../../shared/errors'; +import { AccessError, HiveError } from '../../../shared/errors'; import { Logger } from '../../shared/providers/logger'; -import { TokenStorage } from '../../token/providers/token-storage'; -import { TokensConfig } from '../../token/providers/tokens'; +import { TargetTokenCache } from '../../token/providers/target-token-cache'; import { OrganizationAccessScope, ProjectAccessScope, @@ -69,12 +69,14 @@ export class TargetAccessTokenSession extends Session { export class TargetAccessTokenStrategy extends AuthNStrategy { private logger: Logger; - private tokensConfig: TokensConfig; + private pool: PostgresDatabasePool; + private cache: TargetTokenCache; - constructor(deps: { logger: Logger; tokensConfig: TokensConfig }) { + constructor(deps: { logger: Logger; pool: PostgresDatabasePool; cache: TargetTokenCache }) { super(); this.logger = deps.logger.child({ module: 'TargetAccessTokenStrategy' }); - this.tokensConfig = deps.tokensConfig; + this.pool = deps.pool; + this.cache = deps.cache; } async parse(args: { @@ -124,11 +126,15 @@ export class TargetAccessTokenStrategy extends AuthNStrategy }>; + + constructor( + @Inject(REDIS_INSTANCE) redis: Redis, + private pool: PostgresDatabasePool, + prometheusConfig: PrometheusConfig, + ) { + this.cache = new BentoCache({ + default: 'targetTokens', + plugins: prometheusConfig.isEnabled + ? [ + prometheusPlugin({ + prefix: 'bentocache_target_tokens', + }), + ] + : undefined, + stores: { + targetTokens: bentostore() + .useL1Layer( + memoryDriver({ + maxItems: 10_000, + prefix: 'bentocache:target-tokens', + }), + ) + .useL2Layer(redisDriver({ connection: redis, prefix: 'bentocache:target-tokens' })), + }, + }); + } + + async get(token: string) { + const hashedToken = hashTargetToken(token); + + return await this.cache + .getOrSet({ + key: hashedToken, + factory: () => TargetTokenStorage.findByHash({ pool: this.pool })(hashedToken), + ttl: '5min', + grace: '24h', + }) + .then(result => { + if (result) { + void TargetTokenStorage.touchTokenByHash({ pool: this.pool })(hashedToken).catch( + () => {}, + ); + } + return result; + }); + } + + async add(record: Token) { + return await this.cache.set({ + key: record.token, + value: record, + ttl: '5min', + grace: '24h', + }); + } + + async purge(tokens: readonly string[]) { + await Promise.all(tokens.map(token => this.cache.delete({ key: token }))); + } +} diff --git a/packages/services/api/src/modules/token/providers/target-token-storage.ts b/packages/services/api/src/modules/token/providers/target-token-storage.ts new file mode 100644 index 00000000000..0c52b3026f4 --- /dev/null +++ b/packages/services/api/src/modules/token/providers/target-token-storage.ts @@ -0,0 +1,141 @@ +import { createHash } from 'node:crypto'; +import { Injectable, Scope } from 'graphql-modules'; +import { z } from 'zod'; +import { PostgresDatabasePool, psql, type CommonQueryMethods } from '@hive/postgres'; + +const TokenModel = z.object({ + token: z.string(), + tokenAlias: z.string(), + name: z.string(), + date: z.string(), + lastUsedAt: z.string().nullable(), + organization: z.string(), + project: z.string(), + target: z.string(), + scopes: z.array(z.string()), +}); + +const tokenFields = psql` + "token" + , "token_alias" AS "tokenAlias" + , "name" + , to_json("created_at") AS "date" + , to_json("last_used_at") AS "lastUsedAt" + , "organization_id" AS "organization" + , "project_id" AS "project" + , "target_id" AS "target" + , COALESCE("scopes", ARRAY[]::text[]) AS "scopes" +`; + +export function hashTargetToken(token: string) { + return createHash('sha256').update(token).digest('hex'); +} + +@Injectable({ + scope: Scope.Operation, + global: true, +}) +export class TargetTokenStorage { + constructor(private pool: PostgresDatabasePool) {} + + async createToken(args: Omit, 'date' | 'lastUsedAt'>) { + return await this.pool + .one( + psql` + INSERT INTO "tokens" ( + "name" + , "token" + , "token_alias" + , "target_id" + , "project_id" + , "organization_id" + , "scopes" + ) + VALUES ( + ${args.name} + , ${args.token} + , ${args.tokenAlias} + , ${args.target} + , ${args.project} + , ${args.organization} + , ${psql.array(args.scopes, 'text')} + ) + RETURNING ${tokenFields} + `, + ) + .then(TokenModel.parse); + } + + async getTokens(args: { targetId: string }) { + return await this.pool + .any( + psql` + SELECT ${tokenFields} + FROM "tokens" + WHERE + "target_id" = ${args.targetId} + AND "deleted_at" IS NULL + ORDER BY "created_at" DESC + `, + ) + .then(z.array(TokenModel).parse); + } + + async deleteTokens(args: { targetId: string; tokens: readonly string[] }) { + if (args.tokens.length === 0) { + return []; + } + + return await this.pool + .anyFirst( + psql` + UPDATE "tokens" + SET "deleted_at" = NOW() + WHERE + "target_id" = ${args.targetId} + AND "token" = ANY(${psql.array(args.tokens, 'text')}) + RETURNING "token" + `, + ) + .then(z.array(z.string()).parse); + } + + async getTokenBySecret(token: string) { + return await TargetTokenStorage.getTokenBySecret({ pool: this.pool })(token); + } + + static getTokenBySecret(deps: { pool: CommonQueryMethods }) { + return async function getTokenBySecret(token: string) { + return await TargetTokenStorage.findByHash(deps)(hashTargetToken(token)); + }; + } + + static findByHash(deps: { pool: CommonQueryMethods }) { + return async function findByHash(hashedToken: string) { + return await deps.pool + .maybeOne( + psql` + SELECT ${tokenFields} + FROM "tokens" + WHERE + "token" = ${hashedToken} + AND "deleted_at" IS NULL + LIMIT 1 + `, + ) + .then(TokenModel.nullable().parse); + }; + } + + static touchTokenByHash(deps: { pool: CommonQueryMethods }) { + return async function touchTokenBySecret(token: string) { + await deps.pool.query(psql` + UPDATE "tokens" + SET "last_used_at" = NOW() + WHERE + "token" = ${token} + AND "deleted_at" IS NULL + `); + }; + } +} diff --git a/packages/services/api/src/modules/token/providers/token-storage.ts b/packages/services/api/src/modules/token/providers/token-storage.ts index 5c9c5095f09..ab7179b8fc7 100644 --- a/packages/services/api/src/modules/token/providers/token-storage.ts +++ b/packages/services/api/src/modules/token/providers/token-storage.ts @@ -1,7 +1,6 @@ -import { CONTEXT, Inject, Injectable, Scope } from 'graphql-modules'; +import { createHash, randomBytes } from 'node:crypto'; +import { Injectable, Scope } from 'graphql-modules'; import { maskToken } from '@hive/service-common'; -import type { TokensApi } from '@hive/tokens'; -import { createTRPCProxyClient, httpLink } from '@trpc/client'; import type { Token } from '../../../shared/entities'; import { HiveError } from '../../../shared/errors'; import { atomic } from '../../../shared/helpers'; @@ -12,8 +11,8 @@ import type { } from '../../auth/providers/scopes'; import { Logger } from '../../shared/providers/logger'; import type { TargetSelector } from '../../shared/providers/storage'; -import type { TokensConfig } from './tokens'; -import { TOKENS_CONFIG } from './tokens'; +import { TargetTokenCache } from './target-token-cache'; +import { TargetTokenStorage } from './target-token-storage'; export interface TokenSelector { token: string; @@ -34,31 +33,22 @@ export interface CreateTokenResult extends Token { }) export class TokenStorage { private logger: Logger; - private tokensService; constructor( logger: Logger, - @Inject(TOKENS_CONFIG) tokensConfig: TokensConfig, - @Inject(CONTEXT) context: GraphQLModules.ModuleContext, + private targetTokenStorage: TargetTokenStorage, + private targetTokenCache: TargetTokenCache, ) { this.logger = logger.child({ source: 'TokenStorage' }); - this.tokensService = createTRPCProxyClient({ - links: [ - httpLink({ - url: `${tokensConfig.endpoint}/trpc`, - fetch, - headers: { - 'x-request-id': context.requestId, - }, - }), - ], - }); } async createToken(input: CreateTokenInput) { this.logger.debug('Creating new token (input=%o)', input); - const response = await this.tokensService.createToken.mutate({ + const secret = randomBytes(16).toString('hex'); + const token = await this.targetTokenStorage.createToken({ + token: createTokenHash(secret), + tokenAlias: maskToken(secret), name: input.name, target: input.targetId, project: input.projectId, @@ -66,7 +56,9 @@ export class TokenStorage { scopes: input.scopes, }); - return response; + await this.targetTokenCache.add(token); + + return { ...token, secret }; } async deleteTokens(input: { @@ -75,59 +67,37 @@ export class TokenStorage { }): Promise { this.logger.debug('Deleting tokens (input=%o)', input); - const deletedIds: Array = []; - - await Promise.all( - input.tokenIds.map(token => - this.tokensService.deleteToken - .mutate({ - targetId: input.targetId, - token, - }) - .then(didDelete => { - if (!didDelete) { - return; - } - deletedIds.push(token); - }), - ), - ); - - return deletedIds; + const deletedTokens = await this.targetTokenStorage.deleteTokens({ + targetId: input.targetId, + tokens: input.tokenIds, + }); + await this.targetTokenCache.purge(deletedTokens); + + return deletedTokens; } async invalidateTokens(tokens: string[]) { this.logger.debug('Invalidating tokens (size=%s)', tokens.length); - await this.tokensService.invalidateTokens - .mutate({ - tokens, - }) - .catch(error => { - this.logger.error(error); - }); + await this.targetTokenCache.purge(tokens); } async getTokens(selector: TargetSelector) { this.logger.debug('Fetching tokens (selector=%o)', selector); - const response = await this.tokensService.targetTokens.query({ - targetId: selector.targetId, - }); - - return response || []; + return await this.targetTokenStorage.getTokens({ targetId: selector.targetId }); } @atomic(({ token }) => token) async getToken({ token }: TokenSelector) { try { - // Tokens are MD5 hashes, so they are always 32 characters long + // Target-token secrets are 16 random bytes encoded as hexadecimal. if (token.length !== 32) { throw new HiveError(`Incorrect length: received ${token.length}, expected 32`); } this.logger.debug('Fetching token (token=%s)', maskToken(token)); - const tokenInfo = await this.tokensService.getToken.query({ token }); + const tokenInfo = await this.targetTokenCache.get(token); if (!tokenInfo) { throw new HiveError('Not found'); @@ -149,3 +119,7 @@ export class TokenStorage { } } } + +function createTokenHash(token: string) { + return createHash('sha256').update(token).digest('hex'); +} diff --git a/packages/services/api/src/modules/token/providers/tokens.ts b/packages/services/api/src/modules/token/providers/tokens.ts deleted file mode 100644 index ba045fd8f18..00000000000 --- a/packages/services/api/src/modules/token/providers/tokens.ts +++ /dev/null @@ -1,7 +0,0 @@ -import { InjectionToken } from 'graphql-modules'; - -export interface TokensConfig { - endpoint: string; -} - -export const TOKENS_CONFIG = new InjectionToken('tokens-endpoint'); diff --git a/packages/services/server/.env.template b/packages/services/server/.env.template index c0772403121..999c52e33c3 100644 --- a/packages/services/server/.env.template +++ b/packages/services/server/.env.template @@ -13,7 +13,6 @@ CLICKHOUSE_HOST="localhost" CLICKHOUSE_PORT="8123" CLICKHOUSE_USERNAME="test" CLICKHOUSE_PASSWORD="test" -TOKENS_ENDPOINT="http://localhost:6001" SCHEMA_ENDPOINT="http://localhost:6500" SCHEMA_POLICY_ENDPOINT="http://localhost:6600" COMMERCE_ENDPOINT="http://localhost:4013" diff --git a/packages/services/server/README.md b/packages/services/server/README.md index 46f57000e57..8815820a563 100644 --- a/packages/services/server/README.md +++ b/packages/services/server/README.md @@ -10,7 +10,6 @@ The GraphQL API for GraphQL Hive. | `ENCRYPTION_SECRET` | **Yes** | Secret for encrypting stuff. | `8ebe95cg21c1fee355e9fa32c8c33141` | | `WEB_APP_URL` | **Yes** | The url of the web app. | `http://127.0.0.1:3000` | | `GRAPHQL_PUBLIC_ORIGIN` | **Yes** | The origin of the GraphQL server. | `http://127.0.0.1:4013` | -| `TOKENS_ENDPOINT` | **Yes** | The endpoint of the tokens service. | `http://127.0.0.1:6001` | | `SCHEMA_ENDPOINT` | **Yes** | The endpoint of the schema service. | `http://127.0.0.1:6500` | | `SCHEMA_POLICY_ENDPOINT` | **No** | The endpoint of the schema policy service. | `http://127.0.0.1:6600` | | `POSTGRES_SSL` | No | Whether the postgres connection should be established via SSL. | `1` (enabled) or `0` (disabled) | diff --git a/packages/services/server/src/environment.ts b/packages/services/server/src/environment.ts index 7277bcc0ebd..99d19178eea 100644 --- a/packages/services/server/src/environment.ts +++ b/packages/services/server/src/environment.ts @@ -45,7 +45,6 @@ const EnvironmentModel = zod.object({ }) .url(), SCHEMA_POLICY_ENDPOINT: emptyString(zod.string().url().optional()), - TOKENS_ENDPOINT: zod.string().url(), SCHEMA_ENDPOINT: zod.string().url(), AUTH_ORGANIZATION_OIDC: emptyString(zod.union([zod.literal('1'), zod.literal('0')]).optional()), AUTH_REQUIRE_EMAIL_VERIFICATION: emptyString( @@ -419,9 +418,6 @@ export const env = { webApp: { url: base.WEB_APP_URL, }, - tokens: { - endpoint: base.TOKENS_ENDPOINT, - }, commerce: commerce.COMMERCE_ENDPOINT ? { endpoint: commerce.COMMERCE_ENDPOINT, diff --git a/packages/services/server/src/index.ts b/packages/services/server/src/index.ts index 6433d36846b..5df63315fd4 100644 --- a/packages/services/server/src/index.ts +++ b/packages/services/server/src/index.ts @@ -48,6 +48,7 @@ import { SuperTokensUserAuthNStrategy } from '../../api/src/modules/auth/lib/sup import { TargetAccessTokenStrategy } from '../../api/src/modules/auth/lib/target-access-token-strategy'; import { OrganizationAccessTokenValidationCache } from '../../api/src/modules/auth/providers/organization-access-token-validation-cache'; import { OrganizationAccessTokensCache } from '../../api/src/modules/organization/providers/organization-access-tokens-cache'; +import { TargetTokenCache } from '../../api/src/modules/token/providers/target-token-cache'; import { internalApiRouter } from './api'; import { asyncStorage } from './async-storage'; import { env } from './environment'; @@ -268,9 +269,6 @@ export async function main() { rateLimit: env.supertokens.rateLimit, } : null, - tokens: { - endpoint: env.hiveServices.tokens.endpoint, - }, commerce: { endpoint: env.hiveServices.commerce ? env.hiveServices.commerce.endpoint : null, billingEnabled: env.hiveServices.commerce ? env.hiveServices.commerce.billing : false, @@ -397,9 +395,8 @@ export async function main() { (logger: Logger) => new TargetAccessTokenStrategy({ logger, - tokensConfig: { - endpoint: env.hiveServices.tokens.endpoint, - }, + pool: storage.pool, + cache: registry.injector.get(TargetTokenCache), }), ], }), diff --git a/packages/services/storage/src/index.ts b/packages/services/storage/src/index.ts index a1194a60c9c..5bdbded5a4a 100644 --- a/packages/services/storage/src/index.ts +++ b/packages/services/storage/src/index.ts @@ -33,7 +33,6 @@ import { export type { Interceptor }; -export { createTokenStorage } from './tokens'; export type { tokens, schema_policy_resource } from './db/types'; const organizationGetStartedMapping: Record< diff --git a/packages/services/storage/src/tokens.ts b/packages/services/storage/src/tokens.ts deleted file mode 100644 index 319a2eb7141..00000000000 --- a/packages/services/storage/src/tokens.ts +++ /dev/null @@ -1,151 +0,0 @@ -import { z } from 'zod'; -import { createPostgresDatabasePool, psql, toDate, type Interceptor } from '@hive/postgres'; - -const TokenModel = z.object({ - token: z.string(), - tokenAlias: z.string(), - name: z.string(), - date: z.string(), - lastUsedAt: z.string().nullable(), - organization: z.string(), - project: z.string(), - target: z.string(), - scopes: z - .array(z.string()) - .nullable() - .transform(value => value ?? []), -}); - -const tokenFields = psql` - "token" - , "token_alias" AS "tokenAlias" - , "name" - , to_json("created_at") AS "date" - , to_json("last_used_at") AS "lastUsedAt" - , "organization_id" AS "organization" - , "project_id" AS "project" - , "target_id" AS "target" - , "scopes" -`; - -export async function createTokenStorage( - connection: string, - maximumPoolSize: number, - additionalInterceptors: Interceptor[] = [], -) { - const pool = await createPostgresDatabasePool({ - connectionParameters: connection, - maximumPoolSize, - additionalInterceptors, - }); - - return { - destroy() { - return pool.end(); - }, - async isReady() { - try { - await pool.exists(psql`SELECT 1`); - return true; - } catch { - return false; - } - }, - async getTokens({ target }: { target: string }) { - return await pool - .any( - psql` - SELECT ${tokenFields} - FROM tokens - WHERE - target_id = ${target} - AND deleted_at IS NULL - ORDER BY created_at DESC - `, - ) - .then(z.array(TokenModel).parse); - }, - async getToken({ token }: { token: string }) { - return await pool - .maybeOne( - psql` - SELECT ${tokenFields} - FROM tokens - WHERE token = ${token} AND deleted_at IS NULL - LIMIT 1 - `, - ) - .then(TokenModel.nullable().parse); - }, - async createToken({ - token, - tokenAlias, - target, - project, - organization, - name, - scopes, - }: { - token: string; - tokenAlias: string; - name: string; - target: string; - project: string; - organization: string; - scopes: readonly string[]; - }) { - return await pool - .one( - psql` - INSERT INTO tokens - (name, token, token_alias, target_id, project_id, organization_id, scopes) - VALUES - (${name}, ${token}, ${tokenAlias}, ${target}, ${project}, ${organization}, ${psql.array( - scopes, - 'text', - )}) - RETURNING ${tokenFields} - `, - ) - .then(TokenModel.parse); - }, - async deleteToken(params: { - targetId: string; - token: string; - postDeletionTransaction: () => Promise; - }) { - return await pool.transaction('deleteToken', async t => { - const deleted = await t.maybeOneFirst(psql` - UPDATE - "tokens" - SET - "deleted_at" = NOW() - WHERE - "target_id" = ${params.targetId} - AND "token" = ${params.token} - RETURNING true - `); - - if (deleted) { - await params.postDeletionTransaction(); - } - - return !!deleted; - }); - }, - async touchTokens({ tokens }: { tokens: Array<{ token: string; date: Date }> }) { - await pool.query(psql` - UPDATE tokens as t - SET last_used_at = c.last_used_at - FROM ( - VALUES - (${psql.join( - tokens.map(t => psql`${t.token}, ${toDate(t.date)}`), - psql.fragment`), (`, - )}) - ) as c(token, last_used_at) - WHERE c.token = t.token; - `); - }, - }; -} diff --git a/packages/services/tokens/.env.template b/packages/services/tokens/.env.template deleted file mode 100644 index 2336b826f44..00000000000 --- a/packages/services/tokens/.env.template +++ /dev/null @@ -1,14 +0,0 @@ -POSTGRES_USER=postgres -POSTGRES_PASSWORD=postgres -POSTGRES_HOST=localhost -POSTGRES_PORT=5432 -POSTGRES_DB=registry -REDIS_HOST="localhost" -REDIS_PORT="6379" -REDIS_PASSWORD="" -PORT=6001 -SERVER_HOST=:: -SERVER_HOST_IPV6_ONLY=0 -OPENTELEMETRY_COLLECTOR_ENDPOINT="" -LOG_LEVEL="debug" -OPENTELEMETRY_TRACE_USAGE_REQUESTS=1 diff --git a/packages/services/tokens/.gitignore b/packages/services/tokens/.gitignore deleted file mode 100644 index 4c9d7c35a40..00000000000 --- a/packages/services/tokens/.gitignore +++ /dev/null @@ -1,4 +0,0 @@ -*.log -.DS_Store -node_modules -dist diff --git a/packages/services/tokens/LICENSE b/packages/services/tokens/LICENSE deleted file mode 100644 index 1cf5b9c7d23..00000000000 --- a/packages/services/tokens/LICENSE +++ /dev/null @@ -1,21 +0,0 @@ -MIT License - -Copyright (c) 2022 The Guild - -Permission is hereby granted, free of charge, to any person obtaining a copy -of this software and associated documentation files (the "Software"), to deal -in the Software without restriction, including without limitation the rights -to use, copy, modify, merge, publish, distribute, sublicense, and/or sell -copies of the Software, and to permit persons to whom the Software is -furnished to do so, subject to the following conditions: - -The above copyright notice and this permission notice shall be included in all -copies or substantial portions of the Software. - -THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR -IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, -FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE -AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER -LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, -OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE -SOFTWARE. diff --git a/packages/services/tokens/README.md b/packages/services/tokens/README.md deleted file mode 100644 index 961c68fe631..00000000000 --- a/packages/services/tokens/README.md +++ /dev/null @@ -1,36 +0,0 @@ -# `@hive/tokens` - -This service takes care of validating and issuing tokens used for accessing the public facing hive -APIs (usage service and GraphQL API). - -## Configuration - -| Name | Required | Description | Example Value | -| ------------------------------------ | -------- | -------------------------------------------------------------------------------------------------------- | ---------------------------------------------------- | -| `PORT` | **Yes** | The port this service is running on. | `6001` | -| `POSTGRES_HOST` | **Yes** | Host of the postgres database | `127.0.0.1` | -| `POSTGRES_PORT` | **Yes** | Port of the postgres database | `5432` | -| `POSTGRES_DB` | **Yes** | Name of the postgres database. | `registry` | -| `POSTGRES_USER` | **Yes** | User name for accessing the postgres database. | `postgres` | -| `POSTGRES_PASSWORD` | No | Password for accessing the postgres database. | `postgres` | -| `POSTGRES_SSL` | No | Whether the postgres connection should be established via SSL. | `1` (enabled) or `0` (disabled) | -| `REDIS_HOST` | **Yes** | The host of your redis instance. | `"127.0.0.1"` | -| `REDIS_PORT` | **Yes** | The port of your redis instance. | `6379` | -| `REDIS_PASSWORD` | No | The password of your redis instance. | `"apollorocks"` | -| `REDIS_TLS_ENABLED` | **No** | Enable TLS for redis connection (rediss://). | `"0"` | -| `REDIS_USERNAME` | No | The username for Redis Access Control List authentication. | `"default"` | -| `REDIS_CLUSTER_MODE_ENABLED` | No | Enable Redis Cluster mode. | `1` (enabled) or `0` (disabled) | -| `AWS_REGION` | No | The global AWS region for the service. Used as the default region for AWS connections. | `us-east-1` | -| `REDIS_AWS_IAM_AUTH_ENABLED` | No | Enable AWS IAM authentication for Redis (requires `AWS_REGION`). | `1` (enabled) or `0` (disabled) | -| `REDIS_AWS_REGION` | No | AWS region for IAM auth. Falls back to `AWS_REGION`. | `us-east-1` | -| `REDIS_AWS_IAM_CACHE_NAME` | No | ElastiCache cache name (required if `REDIS_AWS_IAM_AUTH_ENABLED` is `1`). | `my-cache` | -| `ENVIRONMENT` | No | The environment of your Hive app. (**Note:** This will be used for Sentry reporting.) | `staging` | -| `SENTRY` | No | Whether Sentry error reporting should be enabled. | `1` (enabled) or `0` (disabled) | -| `SENTRY_DSN` | No | The DSN for reporting errors to Sentry. | `https://dooobars@o557896.ingest.sentry.io/12121212` | -| `PROMETHEUS_METRICS` | No | Whether Prometheus metrics should be enabled | `1` (enabled) or `0` (disabled) | -| `PROMETHEUS_METRICS_LABEL_INSTANCE` | No | The instance label added for the prometheus metrics. | `tokens` | -| `PROMETHEUS_METRICS_PORT` | No | Port on which prometheus metrics are exposed | Defaults to `10254` | -| `REQUEST_LOGGING` | No | Log http requests | `1` (enabled) or `0` (disabled) | -| `LOG_LEVEL` | No | The verbosity of the service logs. One of `trace`, `debug`, `info`, `warn` ,`error`, `fatal` or `silent` | `info` (default) | -| `OPENTELEMETRY_COLLECTOR_ENDPOINT` | No | OpenTelemetry Collector endpoint. The expected traces transport is HTTP (port `4318`). | `http://localhost:4318/v1/traces` | -| `OPENTELEMETRY_TRACE_USAGE_REQUESTS` | No | If enabled, requests send to this service from `usage` service will be monitored with OTEL. | `1` (enabled, or ``) | diff --git a/packages/services/tokens/package.json b/packages/services/tokens/package.json deleted file mode 100644 index 944b74b7d2f..00000000000 --- a/packages/services/tokens/package.json +++ /dev/null @@ -1,37 +0,0 @@ -{ - "name": "@hive/tokens", - "type": "module", - "license": "MIT", - "private": true, - "exports": { - ".": "./src/api.ts" - }, - "scripts": { - "build": "tsx ../../../scripts/runify.ts", - "dev": "tsup-node --config ../../../configs/tsup/dev.config.node.ts src/dev.ts", - "typecheck": "tsc --noEmit" - }, - "devDependencies": { - "@hive/postgres": "workspace:*", - "@hive/service-common": "workspace:*", - "@hive/storage": "workspace:*", - "@sentry/node": "7.120.2", - "@trpc/server": "10.45.3", - "@types/ms": "0.7.34", - "dotenv": "16.4.7", - "fastify": "5.8.5", - "lru-cache": "11.0.2", - "ms": "2.1.3", - "p-timeout": "6.1.4", - "pino-pretty": "11.3.0", - "reflect-metadata": "0.2.2", - "tiny-lru": "11.4.7", - "tslib": "2.8.1", - "zod": "3.25.76" - }, - "buildOptions": { - "external": [ - "pg-native" - ] - } -} diff --git a/packages/services/tokens/src/api.ts b/packages/services/tokens/src/api.ts deleted file mode 100644 index dea440c7066..00000000000 --- a/packages/services/tokens/src/api.ts +++ /dev/null @@ -1,201 +0,0 @@ -import { createHash } from 'node:crypto'; -import type { FastifyRequest } from 'fastify'; -import { LRU as LruType } from 'tiny-lru'; -import { z } from 'zod'; -import { createErrorHandler, handleTRPCError, maskToken, metrics } from '@hive/service-common'; -import type { inferRouterInputs, inferRouterOutputs } from '@trpc/server'; -import { initTRPC } from '@trpc/server'; -import { recordTokenRead } from './metrics'; -import { Storage } from './multi-tier-storage'; - -const httpRequests = new metrics.Counter({ - name: 'tokens_http_requests', - help: 'Number of http requests', - labelNames: ['path'], -}); - -const httpRequestDuration = new metrics.Histogram({ - name: 'tokens_http_request_duration_seconds', - help: 'Duration of an http request', - labelNames: ['path'], -}); - -function hashToken(token: string) { - return createHash('sha256').update(token).digest('hex'); -} - -export function generateToken() { - const token = createHash('md5') - .update(String(Math.random())) - .update(String(Date.now())) - .digest('hex'); - - const hash = hashToken(token); - const alias = maskToken(token); - - return { - secret: token, - hash, - alias, - }; -} - -export type Context = { - req: FastifyRequest; - errorHandler: ReturnType; - storage: Storage; - tokenReadFailuresCache: LruType; -}; - -const t = initTRPC.context().create(); - -const prometheusMiddleware = t.middleware(async ({ next, path }) => { - const stopTimer = httpRequestDuration.startTimer({ path }); - httpRequests.inc({ path }); - try { - return await next(); - } finally { - stopTimer(); - } -}); - -const procedure = t.procedure.use(prometheusMiddleware).use(handleTRPCError); - -export const tokensApiRouter = t.router({ - targetTokens: procedure - .input( - z - .object({ - targetId: z.string().min(1), - }) - .required(), - ) - .query(async ({ ctx, input }) => { - try { - return await ctx.storage.readTarget(input.targetId); - } catch (error) { - ctx.errorHandler('Failed to get tokens of a target', error as Error); - - throw error; - } - }), - invalidateTokens: procedure - .input( - z - .object({ - tokens: z.array(z.string().min(1)), - }) - .required(), - ) - .mutation(async ({ ctx, input }) => { - try { - await ctx.storage.invalidateTokens(input.tokens); - - return true; - } catch (error) { - ctx.errorHandler('Failed to invalidate tokens', error as Error); - - throw error; - } - }), - createToken: procedure - .input( - z - .object({ - name: z.string().nonempty(), - target: z.string().nonempty(), - project: z.string().nonempty(), - organization: z.string().nonempty(), - scopes: z.array(z.string().nonempty()), - }) - .required(), - ) - .mutation(async ({ ctx, input }) => { - try { - const { target, project, organization, name, scopes } = input; - const token = generateToken(); - const result = await ctx.storage.writeToken({ - name, - target, - project, - organization, - scopes, - token: token.hash, - tokenAlias: token.alias, - }); - - return { - ...result, - secret: token.secret, - }; - } catch (error) { - ctx.errorHandler('Failed to create a token', error as Error); - - throw error; - } - }), - deleteToken: procedure - .input( - z - .object({ - token: z.string().min(1), - targetId: z.string().uuid(), - }) - .required(), - ) - .mutation(async ({ ctx, input }) => { - try { - const result = await ctx.storage.deleteToken({ - targetId: input.targetId, - hashedToken: input.token, - }); - - return result; - } catch (error) { - ctx.errorHandler('Failed to delete a token', error as Error); - - throw error; - } - }), - getToken: procedure - .input( - z - .object({ - token: z.string().length(32, 'token should be 32 characters long'), - }) - .required(), - ) - .query(async ({ ctx, input }) => { - const hash = hashToken(input.token); - const alias = maskToken(input.token); - - // In case the token was not found (or we failed to fetch it) - const cachedFailure = ctx.tokenReadFailuresCache.get(hash); - - if (cachedFailure) { - throw new Error(cachedFailure); - } - - try { - const result = await ctx.storage.readToken(hash, alias); - recordTokenRead(result ? 200 : 404); - // removes the token from the failures cache (in case the value expired) - ctx.tokenReadFailuresCache.delete(hash); - - return result; - } catch (error) { - ctx.errorHandler(`Failed to get a token "${alias}"`, error as Error, ctx.req.log); - - // set token read as failure - // so we don't try to read it again for next X minutes - ctx.tokenReadFailuresCache.set(hash, (error as Error).message); - - recordTokenRead(500); - throw error; - } - }), -}); - -export type TokensApi = typeof tokensApiRouter; -export type TokensApiInput = inferRouterInputs; -export type TokensApiOutput = inferRouterOutputs; diff --git a/packages/services/tokens/src/dev.ts b/packages/services/tokens/src/dev.ts deleted file mode 100644 index 344eecd9c1d..00000000000 --- a/packages/services/tokens/src/dev.ts +++ /dev/null @@ -1,9 +0,0 @@ -import { config } from 'dotenv'; -import 'reflect-metadata'; - -config({ - debug: true, - encoding: 'utf8', -}); - -await import('./index'); diff --git a/packages/services/tokens/src/environment.ts b/packages/services/tokens/src/environment.ts deleted file mode 100644 index 4c677718280..00000000000 --- a/packages/services/tokens/src/environment.ts +++ /dev/null @@ -1,181 +0,0 @@ -import zod from 'zod'; -import { - OpenTelemetryConfigurationModel, - parseRedisConfigFromEnvironment, - resolveServerListenOptions, -} from '@hive/service-common'; - -const isNumberString = (input: unknown) => zod.string().regex(/^\d+$/).safeParse(input).success; - -const numberFromNumberOrNumberString = (input: unknown): number | undefined => { - if (typeof input == 'number') return input; - if (isNumberString(input)) return Number(input); -}; - -const NumberFromString = zod.preprocess(numberFromNumberOrNumberString, zod.number().min(1)); - -// treat an empty string (`''`) as undefined -const emptyString = (input: T) => { - return zod.preprocess((value: unknown) => { - if (value === '') return undefined; - return value; - }, input); -}; - -function raiseInvariant(reason: string): never { - throw new Error(reason); -} - -const EnvironmentModel = zod.object({ - PORT: emptyString(NumberFromString.optional()), - SERVER_HOST: emptyString(zod.string().optional()), - SERVER_HOST_IPV6_ONLY: emptyString(zod.union([zod.literal('1'), zod.literal('0')]).optional()), - ENVIRONMENT: emptyString(zod.string().optional()), - RELEASE: emptyString(zod.string().optional()), - HEARTBEAT_ENDPOINT: emptyString(zod.string().url().optional()), - AWS_REGION: emptyString(zod.string().optional()), -}); - -const SentryModel = zod.union([ - zod.object({ - SENTRY: emptyString(zod.literal('0').optional()), - }), - zod.object({ - SENTRY: zod.literal('1'), - SENTRY_DSN: zod.string(), - }), -]); - -const PostgresModel = zod.object({ - POSTGRES_HOST: zod.string(), - POSTGRES_PORT: NumberFromString, - POSTGRES_PASSWORD: emptyString(zod.string().optional()), - POSTGRES_USER: zod.string(), - POSTGRES_DB: zod.string(), - POSTGRES_SSL: emptyString(zod.union([zod.literal('1'), zod.literal('0')]).optional()), -}); - -const PrometheusModel = zod.object({ - PROMETHEUS_METRICS: emptyString(zod.union([zod.literal('0'), zod.literal('1')]).optional()), - PROMETHEUS_METRICS_LABEL_INSTANCE: emptyString(zod.string().optional()), - PROMETHEUS_METRICS_PORT: emptyString(NumberFromString.optional()), -}); - -const LogModel = zod.object({ - LOG_LEVEL: emptyString( - zod - .union([ - zod.literal('trace'), - zod.literal('debug'), - zod.literal('info'), - zod.literal('warn'), - zod.literal('error'), - zod.literal('fatal'), - zod.literal('silent'), - ]) - .optional(), - ), - REQUEST_LOGGING: emptyString(zod.union([zod.literal('0'), zod.literal('1')]).optional()).default( - '1', - ), -}); - -const configs = { - base: EnvironmentModel.safeParse(process.env), - - sentry: SentryModel.safeParse(process.env), - - postgres: PostgresModel.safeParse(process.env), - - prometheus: PrometheusModel.safeParse(process.env), - - log: LogModel.safeParse(process.env), - tracing: zod - .object({ - ...OpenTelemetryConfigurationModel.shape, - OPENTELEMETRY_TRACE_USAGE_REQUESTS: emptyString(zod.literal('1').optional()), - }) - - .safeParse(process.env), -}; - -const environmentErrors: Array = []; - -for (const config of Object.values(configs)) { - if (config.success === false) { - environmentErrors.push(JSON.stringify(config.error.format(), null, 4)); - } -} - -const redisConfigResult = parseRedisConfigFromEnvironment( - process.env, - configs.base.success ? configs.base.data.AWS_REGION : undefined, -); - -if (redisConfigResult.type === 'error') { - environmentErrors.push(...redisConfigResult.errors); -} - -if (environmentErrors.length) { - const fullError = environmentErrors.join(`\n`); - console.error('❌ Invalid environment variables:', fullError); - process.exit(1); -} - -function extractConfig(config: zod.SafeParseReturnType): Output { - if (!config.success) { - throw new Error('Something went wrong.'); - } - return config.data; -} - -const base = extractConfig(configs.base); -const postgres = extractConfig(configs.postgres); -const sentry = extractConfig(configs.sentry); -const prometheus = extractConfig(configs.prometheus); -const log = extractConfig(configs.log); -const tracing = extractConfig(configs.tracing); - -export const env = { - environment: base.ENVIRONMENT, - release: base.RELEASE ?? 'local', - http: { - port: base.PORT ?? 6001, - ...resolveServerListenOptions({ - serverHost: base.SERVER_HOST, - serverHostIpv6Only: base.SERVER_HOST_IPV6_ONLY, - }), - }, - tracing: { - enabled: !!tracing.OPENTELEMETRY_COLLECTOR_ENDPOINT, - collectorEndpoint: tracing.OPENTELEMETRY_COLLECTOR_ENDPOINT, - traceRequestsFromUsageService: tracing.OPENTELEMETRY_TRACE_USAGE_REQUESTS === '1', - }, - postgres: { - host: postgres.POSTGRES_HOST, - port: postgres.POSTGRES_PORT, - db: postgres.POSTGRES_DB, - user: postgres.POSTGRES_USER, - password: postgres.POSTGRES_PASSWORD, - ssl: postgres.POSTGRES_SSL === '1', - }, - redis: - redisConfigResult?.type === 'ok' - ? redisConfigResult.config - : raiseInvariant('Unreachable: redis config errors are caught above via process.exit(1)'), - heartbeat: base.HEARTBEAT_ENDPOINT ? { endpoint: base.HEARTBEAT_ENDPOINT } : null, - sentry: sentry.SENTRY === '1' ? { dsn: sentry.SENTRY_DSN } : null, - log: { - level: log.LOG_LEVEL ?? 'info', - requests: log.REQUEST_LOGGING === '1', - }, - prometheus: - prometheus.PROMETHEUS_METRICS === '1' - ? { - labels: { - instance: prometheus.PROMETHEUS_METRICS_LABEL_INSTANCE ?? 'tokens', - }, - port: prometheus.PROMETHEUS_METRICS_PORT ?? 10_254, - } - : null, -} as const; diff --git a/packages/services/tokens/src/helpers.ts b/packages/services/tokens/src/helpers.ts deleted file mode 100644 index 385b70d2479..00000000000 --- a/packages/services/tokens/src/helpers.ts +++ /dev/null @@ -1,83 +0,0 @@ -import pTimeout from 'p-timeout'; - -const atomicPromisesInFlight = new Map>(); - -/** - * This function is used to share execution across multiple calls of the same function. - * It's useful when you have a function that can be called multiple times in a short period of time, - * but you want to execute it only once. - * - * Once the execution is finished, the function will be available for the next call. - * - * @param fn - Function that should be executed only once per its execution period. - * @returns Function that will execute the original function only once. - */ -export function atomic(fn: () => Promise): () => Promise { - // Generate a unique string for each call of `atomic` function to prevent collisions. - const uniqueId = Math.random().toString(36).slice(2) + Date.now().toString(36); - - return function atomicWrapper() { - const existing = atomicPromisesInFlight.get(uniqueId); - if (existing) { - return existing; - } - - const promise = fn(); - atomicPromisesInFlight.set(uniqueId, promise); - - return promise.finally(() => { - atomicPromisesInFlight.delete(uniqueId); - }); - }; -} - -/** - * It's used to track the number of requests that are in flight. - * This is important because we don't want to kill the pod when - * state mutating requests are in progress. - */ -export function useActionTracker() { - let actionsInProgress = 0; - - function done() { - --actionsInProgress; - } - - function started() { - ++actionsInProgress; - } - - return { - wrap(fn: (arg: A) => Promise) { - return (arg: A) => { - started(); - return fn(arg).finally(done); - }; - }, - idle() { - return actionsInProgress === 0; - }, - }; -} - -/** - * This function is used to wait until the condition is met or the timeout is reached. - * - * @param conditionFn - function to check the condition - * @param timeout - timeout in milliseconds - */ -export function until(conditionFn: () => boolean, timeout: number): Promise { - return pTimeout( - new Promise(resolve => { - const interval = setInterval(() => { - if (conditionFn()) { - clearInterval(interval); - resolve(); - } - }, 200); - }), - { - milliseconds: timeout, - }, - ); -} diff --git a/packages/services/tokens/src/index.ts b/packages/services/tokens/src/index.ts deleted file mode 100644 index a3872d89612..00000000000 --- a/packages/services/tokens/src/index.ts +++ /dev/null @@ -1,174 +0,0 @@ -#!/usr/bin/env node -import ms from 'ms'; -import 'reflect-metadata'; -import { lru } from 'tiny-lru'; -import { - configureTracing, - createErrorHandler, - createRedisClient, - createServer, - registerShutdown, - registerTRPC, - reportReadiness, - SamplingDecision, - sentryInit, - startHeartbeats, - startMetrics, - TracingInstance, -} from '@hive/service-common'; -import * as Sentry from '@sentry/node'; -import { Context, tokensApiRouter } from './api'; -import { env } from './environment'; -import { createStorage } from './multi-tier-storage'; - -export async function main() { - let tracing: TracingInstance | undefined; - - if (env.tracing.enabled && env.tracing.collectorEndpoint) { - tracing = configureTracing({ - collectorEndpoint: env.tracing.collectorEndpoint, - serviceName: 'tokens', - sampler(ctx, traceId, spanName, spanKind, attributes) { - if (attributes['requesting.service'] === 'usage') { - return { - decision: SamplingDecision.NOT_RECORD, - }; - } - - return { - decision: SamplingDecision.RECORD_AND_SAMPLED, - }; - }, - }); - - tracing.instrumentNodeFetch(); - tracing.setup(); - } - - if (env.sentry) { - sentryInit({ - enabled: true, - environment: env.environment, - dsn: env.sentry.dsn, - release: env.release, - dist: 'tokens', - }); - } - - const server = await createServer({ - name: 'tokens', - sentryErrorHandler: true, - log: { - level: env.log.level, - requests: env.log.requests, - }, - }); - - if (tracing) { - await server.register(...tracing.instrumentFastify()); - } - - const errorHandler = createErrorHandler(server); - - const redis = await createRedisClient(env.redis, { - logger: server.log.child({ source: 'Redis' }), - maxRetriesPerRequest: 20, - }); - - const storage = await createStorage( - env.postgres, - redis, - server.log, - tracing ? [tracing.instrumentSlonik()] : [], - ); - - const stopHeartbeats = env.heartbeat - ? startHeartbeats({ - enabled: true, - endpoint: env.heartbeat.endpoint, - intervalInMS: 20_000, - onError: e => server.log.error(e, `Heartbeat failed with error`), - isReady: storage.isReady, - }) - : startHeartbeats({ enabled: false }); - - async function shutdown() { - stopHeartbeats(); - await server.close(); - await storage.close(); - } - - try { - redis.on('end', async () => { - server.log.info('Redis ended - no more reconnections will be made'); - await shutdown(); - }); - - // Cache failures for 1 minute - const errorCachingInterval = ms('1m'); - const tokenReadFailuresCache = lru(1000, errorCachingInterval); - - registerShutdown({ - logger: server.log, - onShutdown: shutdown, - }); - - await registerTRPC(server, { - router: tokensApiRouter, - createContext({ req }): Context { - return { - req, - errorHandler, - storage, - tokenReadFailuresCache, - }; - }, - }); - - server.route({ - method: ['GET', 'HEAD'], - url: '/_health', - handler(_req, res) { - void res.status(200).send(); - }, - }); - - server.route({ - method: ['GET', 'HEAD'], - url: '/_readiness', - async handler(_, res) { - const isReady = await storage.isReady(); - reportReadiness(isReady); - void res.status(isReady ? 200 : 400).send(); - }, - }); - - if (env.prometheus) { - await startMetrics(env.prometheus.labels.instance, { - port: env.prometheus.port, - host: env.http.host, - ipv6Only: env.http.ipv6Only, - }); - } - - await server.listen({ - port: env.http.port, - host: env.http.host, - ipv6Only: env.http.ipv6Only, - }); - } catch (error) { - server.log.fatal(error); - Sentry.captureException(error, { - level: 'fatal', - }); - process.exit(1); - } -} - -main().catch(err => { - Sentry.captureException(err, { - level: 'fatal', - }); - console.error(err); - process.exit(1); -}); diff --git a/packages/services/tokens/src/metrics.ts b/packages/services/tokens/src/metrics.ts deleted file mode 100644 index ad7f55f5607..00000000000 --- a/packages/services/tokens/src/metrics.ts +++ /dev/null @@ -1,32 +0,0 @@ -import type { LRUCache } from 'lru-cache'; -import { metrics } from '@hive/service-common'; - -const tokenReads = new metrics.Counter({ - name: 'tokens_reads', - help: 'Number of token reads', - labelNames: ['status'], -}); - -const cacheReads = new metrics.Counter({ - name: 'tokens_cache_reads', - help: 'Number of cache reads', - labelNames: ['status'], -}); - -const cacheFills = new metrics.Counter({ - name: 'tokens_cache_fills', - help: 'Number of cache fills', - labelNames: ['source'], -}); - -export function recordCacheRead(status: NonNullable['fetch']>) { - cacheReads.inc({ status }); -} - -export function recordCacheFill(source: 'db' | 'redis-fresh' | 'redis-stale') { - cacheFills.inc({ source }); -} - -export function recordTokenRead(status: 200 | 500 | 404) { - tokenReads.inc({ status }); -} diff --git a/packages/services/tokens/src/multi-tier-storage.ts b/packages/services/tokens/src/multi-tier-storage.ts deleted file mode 100644 index e26c60df2c0..00000000000 --- a/packages/services/tokens/src/multi-tier-storage.ts +++ /dev/null @@ -1,394 +0,0 @@ -import type { FastifyBaseLogger } from 'fastify'; -import { LRUCache } from 'lru-cache'; -import ms from 'ms'; -import { createConnectionString, type PostgresConnectionParamaters } from '@hive/postgres'; -import type { Redis } from '@hive/service-common'; -import { createTokenStorage, type Interceptor } from '@hive/storage'; -import { captureException, captureMessage } from '@sentry/node'; -import { atomic, until, useActionTracker } from './helpers'; -import { recordCacheFill, recordCacheRead } from './metrics'; - -type CacheEntry = StorageItem | 'not-found'; - -interface StorageItem { - token: string; - name: string; - tokenAlias: string; - date: string; - lastUsedAt: string | null; - organization: string; - project: string; - target: string; - scopes: readonly string[]; -} - -export interface Storage { - close(): Promise; - isReady(): Promise; - readTarget(targetId: string): Promise; - readToken(hashedToken: string, maskedToken: string): Promise; - writeToken(item: Omit): Promise; - deleteToken(args: { targetId: string; hashedToken: string }): Promise; - invalidateTokens(hashedTokens: string[]): Promise; -} - -const cacheConfig = { - inMemory: { - maxEntries: 1000, - ttlInMs: ms('5m'), - }, - redis: { - ttlInMs: ms('1h'), - staleTtlInMs: ms('24h'), - }, - tokenTouchIntervalInMs: ms('60s'), -} as const; - -export async function createStorage( - config: PostgresConnectionParamaters, - redis: Redis, - serverLogger: FastifyBaseLogger, - additionalInterceptors: Interceptor[], -): Promise { - const tracker = useActionTracker(); - const connectionString = createConnectionString(config); - const db = await createTokenStorage(connectionString, 5, additionalInterceptors); - const touch = tokenTouchScheduler(serverLogger, async tokens => { - try { - await db.touchTokens({ tokens }); - } catch (error) { - serverLogger.error('Failed to touch tokens (error=%s)', error); - } - }); - const cache = new LRUCache< - string, - CacheEntry, - { - maskedToken: string; - } - >({ - max: cacheConfig.inMemory.maxEntries, - ttl: cacheConfig.inMemory.ttlInMs, - // Allow to return stale data if the fetchMethod is slow - allowStale: false, - // Don't delete the cache entry if the fetchMethod fails - noDeleteOnFetchRejection: true, - // Allow to return stale data if the fetchMethod fails. - // The rejection reason won't be logged though. - allowStaleOnFetchRejection: true, - // If a cache entry is stale or missing, this method is called - // to fill the cache with fresh data. - // This method is called only once per cache key, - // even if multiple requests are waiting for it. - async fetchMethod(hashedToken, _staleEntry, { context }) { - // Nothing fresh in the in-memory cache, let's check Redis - - const logger = serverLogger.child({ maskedToken: context.maskedToken }); - let redisData: string | null = null; - - if (redis.status === 'ready') { - redisData = await redis.get(generateRedisKey(hashedToken)).catch(error => { - handleStorageError({ - logger, - error, - logMsg: 'Failed to read token from Redis', - tier: 'redis', - action: 'fetch', - }); - return null; - }); - } else { - // If redis is not ready, we fallback to the Db. - // This will put more load on the Db, but it won't break the usage reporting. - // It's a temporary state, as fetched value will be written to in-memory cache, - // and to Redis - when it's back online. - logger.warn('Redis is not ready, falling back to Db'); - captureMessage('Redis was not available as secondary cache', 'warning'); - } - - if (redisData) { - recordCacheFill('redis-fresh'); - logger.debug('Returning fresh data from Redis'); - return JSON.parse(redisData) as CacheEntry; - } - - try { - // Nothing in Redis, let's check the DB - const dbResult = await db.getToken({ token: hashedToken }); - const cacheEntry = dbResult ?? 'not-found'; - recordCacheFill('db'); - - // Write to Redis, so the next time we can read it from there - await setInRedis(redis, hashedToken, cacheEntry).catch(error => { - handleStorageError({ - logger, - error, - logMsg: 'Failed to write token to Redis, but it was written to the in-memory cache', - tier: 'redis', - action: 'set', - }); - }); - - return cacheEntry; - } catch (error) { - // If the DB is down, we log the error, and we throw exception. - // This will cause the cache to return stale data. - // This may have a performance impact (more calls to Db), but it won't break the system. - handleStorageError({ - logger, - error, - logMsg: 'Failed to read token from the Db', - tier: 'db', - action: 'fetch', - }); - - if (redis.status !== 'ready') { - logger.warn('Redis is not ready, cannot read stale data from it'); - throw error; - } - - const staleRedisData = await redis.get(generateStaleRedisKey(hashedToken)).catch(error => { - handleStorageError({ - logger, - error, - logMsg: 'Failed to read token from Redis (stale)', - tier: 'redis-stale', - action: 'fetch', - }); - return null; - }); - - if (!staleRedisData) { - logger.debug('No stale data in Redis'); - throw error; - } - - recordCacheFill('redis-stale'); - logger.debug('Returning stale data from Redis'); - // Stale data will be cached in the in-memory cache only, as it's not fresh. - return JSON.parse(staleRedisData) as CacheEntry; - } - - throw new Error('Unexpected code path'); - }, - }); - - return { - async close() { - // Wait for all the pending operations to finish - await until(tracker.idle, 10_000).catch(error => { - serverLogger.error('Failed to wait for tokens being idle', error); - }); - await db.destroy(); - // Wait for Redis to finish all the pending operations - await redis.quit(); - touch.dispose(); - }, - isReady: atomic(async () => { - if (redis.status === 'ready' || redis.status === 'reconnecting') { - return db.isReady(); - } - return false; - }), - async readTarget(target) { - return db.getTokens({ target }); - }, - async readToken(hashedToken, maskedToken) { - const status: LRUCache.Status = {}; - const context = { maskedToken, source: 'in-memory' }; - const tokenLogger = serverLogger.child({ - maskedToken, - }); - const data = await cache.fetch(hashedToken, { - context, - status, - }); - - if (status.fetch) { - recordCacheRead(status.fetch); - } else { - tokenLogger.warn('Status of the fetch is missing'); - } - - if (status.fetchError) { - tokenLogger.error('Fetch error in the In-Memory-Cache (error=%s)', status.fetchError); - } - - if (!data) { - // Looked up in all layers, and the token is not found - return null; - } - - if (data === 'not-found') { - return null; - } - - touch.schedule(hashedToken); - return data; - }, - writeToken: tracker.wrap(async item => { - const result = await db.createToken(item); - - // We don't want to write to in-memory cache, - // as the token may not be used immediately, or at all. - // We write to Redis, so in case the token is used, - // we reuse it for the in-memory cache, without hitting the DB. - const hashedToken = result.token; - const cacheEntry = result; - - // Write to Redis gracefully. If it fails, we log the error, but we don't throw. - // The token won't be in Redis cache, but it will be possible to retrieve it from the DB. - // It will affect performance slightly, but it won't break the system. - try { - await setInRedis(redis, hashedToken, cacheEntry); - } catch (error) { - handleStorageError({ - logger: serverLogger, - error, - logMsg: 'Failed to write token to Redis, but it was created in the Db', - tier: 'redis', - action: 'set', - }); - } - - return cacheEntry; - }), - deleteToken: tracker.wrap(async args => { - return await db.deleteToken({ - targetId: args.targetId, - token: args.hashedToken, - async postDeletionTransaction() { - // delete from Redis when the token is deleted from the DB - await redis.del([ - generateRedisKey(args.hashedToken), - generateStaleRedisKey(args.hashedToken), - ]); - // only then delete from the in-memory cache. - // The other replicas will purge the token from - // their own in-memory caches on their own pace (ttl) - cache.delete(args.hashedToken); - }, - }); - }), - invalidateTokens: tracker.wrap(async hashedTokens => { - if (hashedTokens.length === 0) { - return; - } - - await redis.del( - hashedTokens.map(generateRedisKey).concat(hashedTokens.map(generateStaleRedisKey)), - ); - for (const hashedToken of hashedTokens) { - cache.delete(hashedToken); - } - }), - }; -} - -function generateRedisKey(hashedToken: string) { - // bump the version when the cache format changes - return `tokens:cache:v2:${hashedToken}`; -} - -function generateStaleRedisKey(hashedToken: string) { - // bump the version when the cache format changes - return `tokens:stale-cache:v2:${hashedToken}`; -} - -async function setInRedis(redis: Redis, hashedToken: string, cacheEntry: CacheEntry) { - if (redis.status !== 'ready') { - return; - } - - const stringifiedCacheEntry = JSON.stringify(cacheEntry); - - const results = await redis - .pipeline() - .setex( - generateRedisKey(hashedToken), - Math.ceil(cacheConfig.redis.ttlInMs / 1000), - stringifiedCacheEntry, - ) - .setex( - generateStaleRedisKey(hashedToken), - Math.ceil(cacheConfig.redis.staleTtlInMs / 1000), - stringifiedCacheEntry, - ) - .exec(); - - if (!results?.length) { - return; - } - - const errors: Error[] = []; - - for (const [error] of results) { - if (error instanceof Error) { - errors.push(error); - } - } - - if (errors.length) { - throw new Error(errors.map(e => e.message).join('\n'), { - cause: { - message: 'SETEX Pipeline Failure', - errors, - }, - }); - } -} - -function tokenTouchScheduler( - logger: FastifyBaseLogger, - onTouch: (tokens: Array<{ token: string; date: Date }>) => Promise, -) { - const scheduledTokens = new Map(); - - /** - * Mark token as used - */ - function schedule(hashedToken: string): void { - const now = new Date(); - scheduledTokens.set(hashedToken, now); - } - - const interval = setInterval(() => { - if (!scheduledTokens.size) { - return; - } - - const tokens = Array.from(scheduledTokens.entries()).map(([token, date]) => ({ - token, - date, - })); - scheduledTokens.clear(); - - logger.debug(`Touch ${tokens.length} tokens`); - void onTouch(tokens); - }, cacheConfig.tokenTouchIntervalInMs); - - function dispose() { - clearInterval(interval); - } - - return { - schedule, - dispose, - }; -} - -function handleStorageError(params: { - logger: FastifyBaseLogger; - error: unknown; - logMsg: string; - tier: 'redis' | 'redis-stale' | 'db'; - action: 'fetch' | 'set'; -}) { - params.logger.error(`${params.logMsg} (error=%s)`, params.error); - captureException(params.error, { - tags: { - storageTier: params.tier, - storageAction: params.action, - }, - }); -} diff --git a/packages/services/tokens/tsconfig.json b/packages/services/tokens/tsconfig.json deleted file mode 100644 index 1250a91ab54..00000000000 --- a/packages/services/tokens/tsconfig.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "extends": "../../../tsconfig.json", - "compilerOptions": { - "target": "ES2020", - "module": "esnext", - "rootDir": "../.." - }, - "files": ["src/index.ts"], - "include": ["src/**/*"] -} diff --git a/packages/services/tokens/turbo.json b/packages/services/tokens/turbo.json deleted file mode 100644 index ef90337ae02..00000000000 --- a/packages/services/tokens/turbo.json +++ /dev/null @@ -1,10 +0,0 @@ -{ - "$schema": "https://turborepo.org/schema.json", - "extends": ["//"], - "tasks": { - "dev": { - "persistent": true, - "cache": false - } - } -} diff --git a/packages/services/usage/.env.template b/packages/services/usage/.env.template index 69502fba217..ea752d626f2 100644 --- a/packages/services/usage/.env.template +++ b/packages/services/usage/.env.template @@ -1,4 +1,3 @@ -TOKENS_ENDPOINT="http://localhost:6001" KAFKA_CONNECTION_MODE="docker" KAFKA_BROKER="localhost:9092" KAFKA_BUFFER_SIZE="10" diff --git a/packages/services/usage/README.md b/packages/services/usage/README.md index c9d421ed1b4..2885162ede3 100644 --- a/packages/services/usage/README.md +++ b/packages/services/usage/README.md @@ -10,7 +10,6 @@ The data is written to a Kafka broker, form Kafka the data is feed into clickhou | Name | Required | Description | Example Value | | ----------------------------------- | -------- | ---------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------- | | `PORT` | No | The port this service is running on. | `4001` | -| `TOKENS_ENDPOINT` | **Yes** | The endpoint of the tokens service. | `http://127.0.0.1:6001` | | `COMMERCE_ENDPOINT` | No | The endpoint of the commerce service. | `http://127.0.0.1:4012` | | `KAFKA_TOPIC` | **Yes** | The kafka topic. | `usage_reports_v2` | | `KAFKA_CONSUMER_GROUP` | **Yes** | The kafka consumer group. | `usage_reports_v2` | diff --git a/packages/services/usage/package.json b/packages/services/usage/package.json index 183ba265f44..5c9ddf44908 100644 --- a/packages/services/usage/package.json +++ b/packages/services/usage/package.json @@ -13,12 +13,10 @@ "@hive/postgres": "workspace:*", "@hive/service-common": "workspace:*", "@hive/storage": "workspace:*", - "@hive/tokens": "workspace:*", "@hive/usage-common": "workspace:*", "@sentry/node": "7.120.2", "@sinclair/typebox": "0.34.41", "@trpc/client": "10.45.3", - "@trpc/server": "10.45.3", "ajv": "8.18.0", "dotenv": "16.4.7", "got": "14.4.7", diff --git a/packages/services/usage/src/environment.ts b/packages/services/usage/src/environment.ts index a7b567aaea2..2026827d68b 100644 --- a/packages/services/usage/src/environment.ts +++ b/packages/services/usage/src/environment.ts @@ -31,7 +31,6 @@ const EnvironmentModel = zod.object({ PORT: emptyString(NumberFromString.optional()), SERVER_HOST: emptyString(zod.string().optional()), SERVER_HOST_IPV6_ONLY: emptyString(zod.union([zod.literal('1'), zod.literal('0')]).optional()), - TOKENS_ENDPOINT: zod.string().url(), COMMERCE_ENDPOINT: emptyString(zod.string().url().optional()), RATE_LIMIT_TTL: emptyString(NumberFromString.optional()).default(30_000), ENVIRONMENT: emptyString(zod.string().optional()), @@ -190,9 +189,6 @@ export const env = { collectorEndpoint: tracing.OPENTELEMETRY_COLLECTOR_ENDPOINT, }, hive: { - tokens: { - endpoint: base.TOKENS_ENDPOINT, - }, commerce: base.COMMERCE_ENDPOINT ? { endpoint: base.COMMERCE_ENDPOINT, diff --git a/packages/services/usage/src/index.ts b/packages/services/usage/src/index.ts index c811aea3737..c6f3f8b99c5 100644 --- a/packages/services/usage/src/index.ts +++ b/packages/services/usage/src/index.ts @@ -3,6 +3,7 @@ import 'reflect-metadata'; import { PrometheusConfig } from '@hive/api/modules/shared/providers/prometheus-config'; import { TargetsByIdCache } from '@hive/api/modules/target/providers/targets-by-id-cache'; import { TargetsBySlugCache } from '@hive/api/modules/target/providers/targets-by-slug-cache'; +import { TargetTokenCache } from '@hive/api/modules/token/providers/target-token-cache'; import { createPostgresDatabasePool } from '@hive/postgres'; import { configureTracing, @@ -81,6 +82,7 @@ async function main() { const prometheusConfig = new PrometheusConfig(!!tracing); const targetsByIdCache = new TargetsByIdCache(redis, pgPool, prometheusConfig); const targetsBySlugCache = new TargetsBySlugCache(redis, pgPool, prometheusConfig); + const targetTokenCache = new TargetTokenCache(redis, pgPool, prometheusConfig); if (tracing) { await server.register(...tracing.instrumentFastify()); @@ -117,7 +119,7 @@ async function main() { }); const tokens = createTokens({ - endpoint: env.hive.tokens.endpoint, + cache: targetTokenCache, logger: server.log, }); diff --git a/packages/services/usage/src/metrics.ts b/packages/services/usage/src/metrics.ts index 299013e2330..0cb15910b8a 100644 --- a/packages/services/usage/src/metrics.ts +++ b/packages/services/usage/src/metrics.ts @@ -2,12 +2,12 @@ import { metrics } from '@hive/service-common'; export const tokenRequests = new metrics.Counter({ name: 'usage_tokens_requests', - help: 'Number of requests to Tokens service', + help: 'Number of target token validations', }); export const tokensDuration = new metrics.Histogram({ name: 'usage_tokens_duration_seconds', - help: 'Duration of an HTTP Request to Tokens service in seconds', + help: 'Duration of target token validation in seconds', labelNames: ['status'], buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 3, 4, 5, 7, 10], }); diff --git a/packages/services/usage/src/tokens.ts b/packages/services/usage/src/tokens.ts index be734bf8cae..6498f243fd6 100644 --- a/packages/services/usage/src/tokens.ts +++ b/packages/services/usage/src/tokens.ts @@ -1,7 +1,5 @@ -import { LRUCache } from 'lru-cache'; +import { TargetTokenCache } from '@hive/api/modules/token/providers/target-token-cache'; import { ServiceLogger } from '@hive/service-common'; -import type { TokensApi } from '@hive/tokens'; -import { createTRPCProxyClient, httpLink } from '@trpc/client'; import { tokenRequests } from './metrics'; export enum TokenStatus { @@ -18,56 +16,34 @@ export type TokensResponse = { type Token = TokensResponse | TokenStatus; -export function createTokens(config: { endpoint: string; logger: ServiceLogger }) { - const endpoint = config.endpoint.replace(/\/$/, ''); - const tokensApi = createTRPCProxyClient({ - links: [ - httpLink({ - url: `${endpoint}/trpc`, - fetch, - headers: { - 'x-requesting-service': 'usage', - }, - }), - ], - }); - const tokens = new LRUCache({ - max: 1000, - ttl: 30_000, - allowStale: false, - // If a cache entry is stale or missing, this method is called - // to fill the cache with fresh data. - // This method is called only once per cache key, - // even if multiple requests are waiting for it. - async fetchMethod(token) { - tokenRequests.inc(); - try { - const info = await tokensApi.getToken.query({ - token, - }); +export function createTokens(config: { cache: TargetTokenCache; logger: ServiceLogger }) { + async function fetch(token: string): Promise { + tokenRequests.inc(); - if (info) { - const result = info.scopes.includes('target:registry:write') - ? { - target: info.target, - project: info.project, - organization: info.organization, - scopes: info.scopes, - } - : TokenStatus.NoAccess; - return result; - } - return TokenStatus.NotFound; - } catch (error) { - config.logger.error('Failed to fetch fresh token (error=%s)', error); + try { + const info = await config.cache.get(token); + + if (!info) { return TokenStatus.NotFound; } - }, - }); + + return info.scopes.includes('target:registry:write') + ? { + target: info.target, + project: info.project, + organization: info.organization, + scopes: info.scopes, + } + : TokenStatus.NoAccess; + } catch (error) { + config.logger.error('Failed to fetch target token (error=%s)', error); + return TokenStatus.NotFound; + } + } return { async fetch(token: string) { - return (await tokens.fetch(token)) ?? TokenStatus.NotFound; + return await fetch(token); }, isNotFound(token: Token): token is TokenStatus.NotFound { return token === TokenStatus.NotFound; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index ae11dbc2db5..20e41647d8f 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1121,9 +1121,6 @@ importers: '@hive/storage': specifier: workspace:* version: link:../storage - '@hive/tokens': - specifier: workspace:* - version: link:../tokens '@hive/usage-common': specifier: workspace:* version: link:../usage-common @@ -1838,57 +1835,6 @@ importers: specifier: 3.25.76 version: 3.25.76 - packages/services/tokens: - devDependencies: - '@hive/postgres': - specifier: workspace:* - version: link:../../internal/postgres - '@hive/service-common': - specifier: workspace:* - version: link:../service-common - '@hive/storage': - specifier: workspace:* - version: link:../storage - '@sentry/node': - specifier: 7.120.2 - version: 7.120.2 - '@trpc/server': - specifier: 10.45.3 - version: 10.45.3 - '@types/ms': - specifier: 0.7.34 - version: 0.7.34 - dotenv: - specifier: 16.4.7 - version: 16.4.7 - fastify: - specifier: 5.8.5 - version: 5.8.5 - lru-cache: - specifier: 11.0.2 - version: 11.0.2 - ms: - specifier: 2.1.3 - version: 2.1.3 - p-timeout: - specifier: 6.1.4 - version: 6.1.4 - pino-pretty: - specifier: 11.3.0 - version: 11.3.0 - reflect-metadata: - specifier: 0.2.2 - version: 0.2.2 - tiny-lru: - specifier: 11.4.7 - version: 11.4.7 - tslib: - specifier: 2.8.1 - version: 2.8.1 - zod: - specifier: 3.25.76 - version: 3.25.76 - packages/services/usage: devDependencies: '@hive/api': @@ -1903,9 +1849,6 @@ importers: '@hive/storage': specifier: workspace:* version: link:../storage - '@hive/tokens': - specifier: workspace:* - version: link:../tokens '@hive/usage-common': specifier: workspace:* version: link:../usage-common @@ -1918,9 +1861,6 @@ importers: '@trpc/client': specifier: 10.45.3 version: 10.45.3(@trpc/server@10.45.3) - '@trpc/server': - specifier: 10.45.3 - version: 10.45.3 ajv: specifier: ^8.18.0 version: 8.18.0 diff --git a/tsconfig.json b/tsconfig.json index 8b4f2450176..1a2e09fa412 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -50,7 +50,6 @@ "@hive/usage": ["./packages/services/usage/src/index.ts"], "@hive/usage-ingestor": ["./packages/services/usage-ingestor/src/index.ts"], "@hive/policy": ["./packages/services/policy/src/api.ts"], - "@hive/tokens": ["./packages/services/tokens/src/api.ts"], "@hive/commerce": ["./packages/services/commerce/src/api.ts"], "@hive/storage": ["./packages/services/storage/src/index.ts"], "@hive/storage/*": ["./packages/services/storage/src/*"], diff --git a/turbo.json b/turbo.json index 2da58c4b34a..aeebd814c25 100644 --- a/turbo.json +++ b/turbo.json @@ -25,7 +25,6 @@ "//#graphql:generate:watch", "@hive/server#dev", "@hive/app#dev", - "@hive/tokens#dev", "@hive/schema#dev", "@hive/policy#dev", "@hive/commerce#dev",