智能代码生成与优化专家

代码实现

// package.json
{
  "name": "b2b-saas-authz",
  "version": "1.0.0",
  "type": "module",
  "private": true,
  "scripts": {
    "start": "node --enable-source-maps dist/main.js",
    "start:dev": "nest start --watch",
    "build": "nest build",
    "test": "jest --runInBand",
    "test:cov": "jest --coverage",
    "lint": "eslint . --ext .ts",
    "load": "node scripts/load-test.mjs"
  },
  "dependencies": {
    "@nestjs/common": "^10.4.7",
    "@nestjs/core": "^10.4.7",
    "@nestjs/platform-express": "^10.4.7",
    "class-transformer": "^0.5.1",
    "class-validator": "^0.14.0",
    "helmet": "^7.1.0",
    "ioredis": "^5.4.1",
    "jose": "^5.9.3",
    "lru-cache": "^11.0.1",
    "reflect-metadata": "^0.2.2",
    "rxjs": "^7.8.1",
    "zod": "^3.23.8"
  },
  "devDependencies": {
    "@nestjs/cli": "^10.4.5",
    "@nestjs/schematics": "^10.1.2",
    "@nestjs/testing": "^10.4.7",
    "@types/express": "^4.17.21",
    "@types/jest": "^29.5.12",
    "@types/node": "^20.17.6",
    "@types/supertest": "^6.0.3",
    "autocannon": "^7.14.0",
    "jest": "^29.7.0",
    "supertest": "^7.0.0",
    "ts-jest": "^29.2.5",
    "ts-node": "^10.9.2",
    "typescript": "^5.6.3"
  }
}

// tsconfig.json
{
  "compilerOptions": {
    "module": "ESNext",
    "target": "ES2022",
    "moduleResolution": "Bundler",
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true,
    "strict": true,
    "skipLibCheck": true,
    "sourceMap": true,
    "rootDir": "src",
    "outDir": "dist",
    "esModuleInterop": true
  },
  "include": ["src/**/*.ts", "scripts/**/*.mjs", "test/**/*.ts"]
}

// jest.config.js
module.exports = {
  testEnvironment: 'node',
  transform: { '^.+\\.ts$': ['ts-jest', { tsconfig: 'tsconfig.json' }] },
  moduleFileExtensions: ['ts', 'js', 'json'],
  rootDir: '.',
  testRegex: '.e2e-spec.ts$',
  coverageDirectory: './coverage',
  collectCoverageFrom: ['src/**/*.ts', '!src/main.ts'],
  setupFilesAfterEnv: []
};

// .env.example
PORT=3000
NODE_ENV=development
REDIS_URL=
JWKS_CACHE_TTL_MS=60000
POLICY_CACHE_TTL_MS=60000
PERM_CACHE_TTL_MS=60000
JWKS_ROTATE_INTERVAL_MS=500
AUTH_IAT_SKEW_SEC=60
RATE_LIMIT_PER_MIN=600

// src/main.ts
import 'reflect-metadata';
import { NestFactory } from '@nestjs/core';
import { AppModule } from './app.module.js';
import helmet from 'helmet';
import { ValidationPipe } from '@nestjs/common';
import { GlobalExceptionFilter } from './common/errors/http-exception.filter.js';
import { ConfigService } from './common/config/config.service.js';

async function bootstrap() {
  const app = await NestFactory.create(AppModule, { logger: ['log', 'warn', 'error'] });
  app.use(helmet());
  app.enableShutdownHooks();
  app.useGlobalPipes(new ValidationPipe({ transform: true, whitelist: true, forbidNonWhitelisted: true }));
  app.useGlobalFilters(new GlobalExceptionFilter());
  app.setGlobalPrefix('v1');

  const config = app.get(ConfigService);
  const port = config.get('PORT');
  await app.listen(port);
}
bootstrap();

// src/app.module.ts
import { Module } from '@nestjs/common';
import { ConfigModule } from './common/config/config.module.js';
import { CacheModuleEx } from './common/cache/cache.module.js';
import { AuditModule } from './common/audit/audit.module.js';
import { AuthModule } from './auth/auth.module.js';
import { PolicyModule } from './policy/policy.module.js';
import { ResourceModule } from './resource/resource.module.js';
import { RateLimitModule } from './common/rate-limit/rate-limit.module.js';
import { JwksDevModule } from './dev/jwks-dev.module.js';

@Module({
  imports: [
    ConfigModule,
    CacheModuleEx,
    AuditModule,
    RateLimitModule,
    AuthModule,
    PolicyModule,
    ResourceModule,
    // Dev JWKS server for tests and local
    JwksDevModule
  ],
})
export class AppModule {}

// src/common/config/config.module.ts
import { Module } from '@nestjs/common';
import { ConfigService } from './config.service.js';
@Module({ providers: [ConfigService], exports: [ConfigService] })
export class ConfigModule {}

// src/common/config/config.service.ts
import { Injectable } from '@nestjs/common';
import { z } from 'zod';

const EnvSchema = z.object({
  PORT: z.coerce.number().int().positive().default(3000),
  NODE_ENV: z.enum(['development', 'test', 'production']).default('development'),
  REDIS_URL: z.string().optional().default(''),
  JWKS_CACHE_TTL_MS: z.coerce.number().int().positive().default(60000),
  JWKS_ROTATE_INTERVAL_MS: z.coerce.number().int().positive().default(500),
  POLICY_CACHE_TTL_MS: z.coerce.number().int().positive().default(60000),
  PERM_CACHE_TTL_MS: z.coerce.number().int().positive().default(60000),
  AUTH_IAT_SKEW_SEC: z.coerce.number().int().nonnegative().default(60),
  RATE_LIMIT_PER_MIN: z.coerce.number().int().positive().default(600)
});

export type AppConfig = z.infer<typeof EnvSchema>;

@Injectable()
export class ConfigService {
  private readonly cfg: AppConfig;
  constructor() {
    const parsed = EnvSchema.safeParse(process.env);
    if (!parsed.success) {
      // eslint-disable-next-line no-console
      console.error('Invalid env:', parsed.error.flatten());
      throw new Error('config invalid');
    }
    this.cfg = parsed.data;
  }
  get<K extends keyof AppConfig>(k: K): AppConfig[K] {
    return this.cfg[k];
  }
  all(): AppConfig { return this.cfg; }
}

// src/common/errors/http-exception.filter.ts
import { ArgumentsHost, Catch, ExceptionFilter, HttpException, HttpStatus } from '@nestjs/common';
import { randomUUID } from 'crypto';

@Catch()
export class GlobalExceptionFilter implements ExceptionFilter {
  catch(exception: unknown, host: ArgumentsHost) {
    const ctx = host.switchToHttp();
    const res = ctx.getResponse();
    const req = ctx.getRequest();

    const traceId = req.headers['x-trace-id'] || randomUUID();
    const now = new Date().toISOString();

    let status = HttpStatus.INTERNAL_SERVER_ERROR;
    let message = 'Internal Server Error';
    let code = 'internal_error';
    let details: any = undefined;

    if (exception instanceof HttpException) {
      status = exception.getStatus();
      const resp = exception.getResponse();
      const err = typeof resp === 'string' ? { message: resp } : resp as any;
      message = err.message || message;
      code = err.code || code;
      details = err.details;
    } else if (exception instanceof Error) {
      message = exception.message;
    }

    const body = {
      code,
      message,
      status,
      traceId,
      time: now,
      path: req?.url,
    };
    res.status(status).json({ ...body, details });
  }
}

// src/common/audit/audit.module.ts
import { Module } from '@nestjs/common';
import { AuditService } from './audit.service.js';
@Module({ providers: [AuditService], exports: [AuditService] })
export class AuditModule {}

// src/common/audit/audit.service.ts
import { Injectable } from '@nestjs/common';

export interface AuditEvent {
  ts: string;
  tenantId: string;
  userId: string;
  action: string;
  resource: string;
  allowed: boolean;
  reason?: string;
  ip?: string;
  method?: string;
  path?: string;
  meta?: Record<string, any>;
}

@Injectable()
export class AuditService {
  async logAccess(ev: AuditEvent) {
    // Simple stdout; can be extended to Redis streams/ELK
    // eslint-disable-next-line no-console
    console.log(JSON.stringify({ type: 'audit.access', ...ev }));
  }
}

// src/common/cache/cache.module.ts
import { Module, Global } from '@nestjs/common';
import { CacheService } from './cache.service.js';

@Global()
@Module({
  providers: [CacheService],
  exports: [CacheService],
})
export class CacheModuleEx {}

// src/common/cache/cache.service.ts
import { Injectable, OnModuleDestroy } from '@nestjs/common';
import LRU from 'lru-cache';
import IORedis from 'ioredis';
import { ConfigService } from '../config/config.service.js';

@Injectable()
export class CacheService implements OnModuleDestroy {
  private redis?: IORedis;
  private lru = new LRU<string, any>({ max: 5000, ttlAutopurge: true });

  constructor(private cfg: ConfigService) {
    const url = cfg.get('REDIS_URL');
    if (url) {
      this.redis = new IORedis(url, { lazyConnect: true, maxRetriesPerRequest: 1 });
      this.redis.connect().catch(() => { /* fallback to LRU */ this.redis = undefined; });
    }
  }
  async onModuleDestroy() { if (this.redis) await this.redis.quit(); }

  async get<T>(k: string): Promise<T | undefined> {
    if (this.redis) {
      const v = await this.redis.get(k);
      return v ? JSON.parse(v) as T : undefined;
    }
    return this.lru.get(k) as T | undefined;
  }
  async set<T>(k: string, v: T, ttlMs?: number) {
    if (this.redis) {
      await this.redis.set(k, JSON.stringify(v), ...(ttlMs ? ['PX', ttlMs] : [] as any));
    } else {
      this.lru.set(k, v, { ttl: ttlMs });
    }
  }
  async del(k: string) {
    if (this.redis) await this.redis.del(k);
    else this.lru.delete(k);
  }
  async incrBy(k: string, delta: number, ttlMs?: number): Promise<number> {
    if (this.redis) {
      const val = await this.redis.incrby(k, delta);
      if (ttlMs) await this.redis.pexpire(k, ttlMs);
      return val;
    } else {
      const v = (this.lru.get(k) as number | undefined) ?? 0;
      const nv = v + delta;
      this.lru.set(k, nv, { ttl: ttlMs });
      return nv;
    }
  }
}

// src/common/rate-limit/rate-limit.module.ts
import { Module, Global } from '@nestjs/common';
import { RateLimitGuard } from './rate-limit.guard.js';
import { CacheModuleEx } from '../cache/cache.module.js';
@Global()
@Module({
  imports: [CacheModuleEx],
  providers: [RateLimitGuard],
  exports: [RateLimitGuard]
})
export class RateLimitModule {}

// src/common/rate-limit/rate-limit.guard.ts
import { CanActivate, ExecutionContext, Injectable, HttpException, HttpStatus } from '@nestjs/common';
import { CacheService } from '../cache/cache.service.js';
import { ConfigService } from '../config/config.service.js';

@Injectable()
export class RateLimitGuard implements CanActivate {
  constructor(private cache: CacheService, private cfg: ConfigService) {}
  async canActivate(ctx: ExecutionContext): Promise<boolean> {
    const req = ctx.switchToHttp().getRequest();
    // Key by tenant+user or IP
    const tenantId = req.auth?.tenantId || 'anon';
    const userId = req.auth?.userId || req.ip || 'ip';
    const key = `ratelimit:${tenantId}:${userId}:${new Date().getUTCMinutes()}`;
    const limit = this.cfg.get('RATE_LIMIT_PER_MIN');
    const count = await this.cache.incrBy(key, 1, 60_000);
    if (count > limit) {
      throw new HttpException({ code: 'rate_limit', message: 'Too Many Requests' }, HttpStatus.TOO_MANY_REQUESTS);
    }
    return true;
  }
}

// src/auth/types.ts
export interface AuthContext {
  tenantId: string;
  userId: string;
  roles: string[];
  scopes: string[];
  jti?: string;
  nonce?: string;
  iat?: number;
}
declare global {
  namespace Express {
    interface Request {
      auth?: AuthContext;
    }
  }
}

// src/auth/auth.module.ts
import { Module } from '@nestjs/common';
import { AuthGuard } from './auth.guard.js';
import { JwtService } from './jwt.service.js';
import { RevocationService } from './revocation.service.js';
import { TenantIssuerService } from './tenant-issuer.service.js';
import { CacheModuleEx } from '../common/cache/cache.module.js';
import { ConfigModule } from '../common/config/config.module.js';
import { AuditModule } from '../common/audit/audit.module.js';

@Module({
  imports: [CacheModuleEx, ConfigModule, AuditModule],
  providers: [AuthGuard, JwtService, RevocationService, TenantIssuerService],
  exports: [AuthGuard, JwtService, RevocationService, TenantIssuerService],
})
export class AuthModule {}

// src/auth/tenant-issuer.service.ts
import { Injectable } from '@nestjs/common';

// In production, resolve from DB/config service. Here a static map for demo.
@Injectable()
export class TenantIssuerService {
  private map = new Map<string, { issuer: string; jwksUri: string }>([
    // Example tenants: t1 and t2. JWKs served by our dev module for demo
    ['t1', { issuer: 'https://issuer.local/t1', jwksUri: 'http://localhost:3000/v1/dev/jwks/t1' }],
    ['t2', { issuer: 'https://issuer.local/t2', jwksUri: 'http://localhost:3000/v1/dev/jwks/t2' }],
  ]);

  resolve(tenantId: string) {
    return this.map.get(tenantId);
  }
}

// src/auth/revocation.service.ts
import { Injectable } from '@nestjs/common';
import { CacheService } from '../common/cache/cache.service.js';

// Store revoked jti in cache (Redis or LRU). Key: revoke:<tenant>:<jti> -> 1
@Injectable()
export class RevocationService {
  constructor(private cache: CacheService) {}
  async isRevoked(tenantId: string, jti?: string): Promise<boolean> {
    if (!jti) return false;
    const v = await this.cache.get<number>(`revoke:${tenantId}:${jti}`);
    return v === 1;
  }
  async revoke(tenantId: string, jti: string, expSec: number) {
    await this.cache.set(`revoke:${tenantId}:${jti}`, 1, expSec * 1000);
  }
}

// src/auth/jwt.service.ts
import { Injectable } from '@nestjs/common';
import { createRemoteJWKSet, jwtVerify, JWTPayload, JWSHeaderParameters } from 'jose';
import { URL } from 'url';
import { CacheService } from '../common/cache/cache.service.js';
import { ConfigService } from '../common/config/config.service.js';
import { TenantIssuerService } from './tenant-issuer.service.js';
import type { AuthContext } from './types.js';

@Injectable()
export class JwtService {
  constructor(
    private cache: CacheService,
    private cfg: ConfigService,
    private tenantIssuer: TenantIssuerService
  ) {}

  private jwksCache = new Map<string, ReturnType<typeof createRemoteJWKSet>>();

  private getJwks(jwksUri: string) {
    if (!this.jwksCache.has(jwksUri)) {
      const u = new URL(jwksUri);
      const ttl = this.cfg.get('JWKS_CACHE_TTL_MS');
      const rotate = this.cfg.get('JWKS_ROTATE_INTERVAL_MS');
      const jwks = createRemoteJWKSet(u, { timeoutDuration: 800, cooldownDuration: rotate, cacheMaxAge: ttl });
      this.jwksCache.set(jwksUri, jwks);
    }
    return this.jwksCache.get(jwksUri)!;
  }

  async verifyRS256(token: string, tenantId: string): Promise<{ payload: JWTPayload; header: JWSHeaderParameters }> {
    const issuerInfo = this.tenantIssuer.resolve(tenantId);
    if (!issuerInfo) throw new Error('unknown tenant');
    const audience = `b2b-saas:${tenantId}`;
    const jwks = this.getJwks(issuerInfo.jwksUri);
    const res = await jwtVerify(token, jwks, {
      algorithms: ['RS256'],
      issuer: issuerInfo.issuer,
      audience,
      maxTokenAge: `${this.cfg.get('AUTH_IAT_SKEW_SEC')}s`
    });
    return { payload: res.payload, header: res.protectedHeader };
  }

  toAuthContext(payload: JWTPayload): AuthContext {
    // Standard claims: sub=userId, tid=tenant, roles, scope, jti, nonce
    const roles = Array.isArray(payload['roles']) ? payload['roles'] as string[] : (payload['role'] ? [String(payload['role'])] : []);
    const scopes = typeof payload['scope'] === 'string' ? (payload['scope'] as string).split(' ') : (payload['scopes'] as string[] ?? []);
    return {
      tenantId: String(payload['tid'] ?? ''),
      userId: String(payload.sub ?? ''),
      roles,
      scopes,
      jti: payload.jti,
      nonce: payload['nonce'] as string | undefined,
      iat: payload.iat
    };
  }
}

// src/auth/auth.guard.ts
import { CanActivate, ExecutionContext, HttpException, HttpStatus, Injectable } from '@nestjs/common';
import { JwtService } from './jwt.service.js';
import { RevocationService } from './revocation.service.js';
import { AuditService } from '../common/audit/audit.service.js';
import type { AuthContext } from './types.js';

function parseBearer(h?: string): string | null {
  if (!h) return null;
  const [t, v] = h.split(' ');
  if (t?.toLowerCase() !== 'bearer' || !v) return null;
  return v;
}

@Injectable()
export class AuthGuard implements CanActivate {
  constructor(
    private jwt: JwtService,
    private revoke: RevocationService,
    private audit: AuditService
  ) {}

  async canActivate(ctx: ExecutionContext): Promise<boolean> {
    const req = ctx.switchToHttp().getRequest();
    const token = parseBearer(req.headers['authorization']);
    if (!token) throw new HttpException({ code: 'unauthorized', message: 'Missing token' }, HttpStatus.UNAUTHORIZED);

    // Tenant must be provided via header or route param for lookup
    const tenantId = (req.headers['x-tenant-id'] as string) || req.params?.tenantId || '';
    if (!tenantId) throw new HttpException({ code: 'unauthorized', message: 'Missing tenant' }, HttpStatus.UNAUTHORIZED);

    try {
      const { payload } = await this.jwt.verifyRS256(token, tenantId);
      const auth: AuthContext = this.jwt.toAuthContext(payload);

      if (auth.tenantId !== tenantId) {
        throw new HttpException({ code: 'tenant_mismatch', message: 'Tenant mismatch' }, HttpStatus.FORBIDDEN);
      }

      // Replay protection: nonce optional but if present, enforce single-use via cache
      if (auth.nonce) {
        const cacheKey = `nonce:${tenantId}:${auth.nonce}`;
        // If exists, reject; else store
        // TTL 5 minutes
        const existed = await this.revoke['cache'].get<number>(cacheKey);
        if (existed) throw new HttpException({ code: 'replay', message: 'Nonce replayed' }, HttpStatus.UNAUTHORIZED);
        await this.revoke['cache'].set(cacheKey, 1, 300_000);
      }

      // Revocation check
      if (await this.revoke.isRevoked(tenantId, auth.jti)) {
        throw new HttpException({ code: 'revoked', message: 'Token revoked' }, HttpStatus.UNAUTHORIZED);
      }

      req.auth = auth;
      return true;
    } catch (e: any) {
      await this.audit.logAccess({
        ts: new Date().toISOString(),
        tenantId,
        userId: 'unknown',
        action: 'authenticate',
        resource: 'token',
        allowed: false,
        reason: e?.message,
        ip: req.ip,
        method: req.method,
        path: req.originalUrl
      });
      if (e instanceof HttpException) throw e;
      throw new HttpException({ code: 'unauthorized', message: 'Invalid token' }, HttpStatus.UNAUTHORIZED);
    }
  }
}

// src/policy/types.ts
export type Effect = 'allow' | 'deny';

export interface RolePermission {
  resource: string; // e.g., 'resource', 'policy'
  action: string;   // e.g., 'read', 'write', 'list'
}

export interface RoleDef {
  name: string;
  permissions: RolePermission[];
}

// Condition operators: all/any, eq, in, owner, tenantMatch, timeWindow
export type Condition =
  | { all: Condition[] }
  | { any: Condition[] }
  | { not: Condition }
  | { eq: [path: string, value: string | number | boolean] }
  | { in: [path: string, values: (string | number)[]] }
  | { owner: [path: string] } // path in resource equals auth.userId
  | { tenantMatch: true }
  | { timeWindow: { fromISO?: string; toISO?: string } };

export interface PolicyRule {
  id: string;
  description?: string;
  effect: Effect;
  resource: string;
  action: string;
  condition?: Condition;
}

export interface EvaluationContext {
  auth: {
    tenantId: string;
    userId: string;
    roles: string[];
    scopes: string[];
  };
  resource?: Record<string, any>;
  nowISO?: string;
}

// src/policy/policy.engine.ts
import { Condition, EvaluationContext, PolicyRule } from './types.js';

function getPath(obj: any, path: string): any {
  return path.split('.').reduce((o, k) => (o ? o[k] : undefined), obj);
}

export function evalCondition(cond: Condition, ctx: EvaluationContext): boolean {
  if ('all' in cond) return cond.all.every(c => evalCondition(c, ctx));
  if ('any' in cond) return cond.any.some(c => evalCondition(c, ctx));
  if ('not' in cond) return !evalCondition(cond.not, ctx);
  if ('eq' in cond) {
    const [path, value] = cond.eq;
    return getPath({ ...ctx }, path) === value;
  }
  if ('in' in cond) {
    const [path, values] = cond.in;
    return values.includes(getPath({ ...ctx }, path));
  }
  if ('owner' in cond) {
    const [path] = cond.owner;
    return getPath({ ...ctx }, path) === ctx.auth.userId;
  }
  if ('tenantMatch' in cond) {
    return ctx.resource?.tenantId === ctx.auth.tenantId;
  }
  if ('timeWindow' in cond) {
    const now = ctx.nowISO ? new Date(ctx.nowISO) : new Date();
    const { fromISO, toISO } = cond.timeWindow;
    if (fromISO && now < new Date(fromISO)) return false;
    if (toISO && now > new Date(toISO)) return false;
    return true;
  }
  return false;
}

export function evaluatePolicies(rules: PolicyRule[], ctx: EvaluationContext, resource: string, action: string): { allowed: boolean; matched?: PolicyRule } {
  // Deny by default; explicit deny overrides allow
  let allowMatched: PolicyRule | undefined;
  for (const r of rules) {
    if (r.resource !== resource || r.action !== action) continue;
    const ok = r.condition ? evalCondition(r.condition, { ...ctx }) : true;
    if (!ok) continue;
    if (r.effect === 'deny') return { allowed: false, matched: r };
    if (r.effect === 'allow') allowMatched = r;
  }
  return { allowed: !!allowMatched, matched: allowMatched };
}

// src/policy/policy.service.ts
import { Injectable } from '@nestjs/common';
import { CacheService } from '../common/cache/cache.service.js';
import { ConfigService } from '../common/config/config.service.js';
import { PolicyRule, RoleDef, RolePermission } from './types.js';
import { createHash } from 'crypto';

@Injectable()
export class PolicyService {
  constructor(private cache: CacheService, private cfg: ConfigService) {}

  private roleKey(tenantId: string) { return `roles:${tenantId}`; }
  private polKey(tenantId: string) { return `policies:${tenantId}`; }
  private permCacheKey(tenantId: string, userId: string, roles: string[], scopes: string[]) {
    const tag = createHash('sha1').update(JSON.stringify({ roles, scopes })).digest('hex').slice(0, 8);
    return `perm:${tenantId}:${userId}:${tag}`;
  }

  async getRoles(tenantId: string): Promise<RoleDef[]> {
    return (await this.cache.get<RoleDef[]>(this.roleKey(tenantId))) ?? [];
    // In production, fallback to DB if not in cache.
  }
  async setRoles(tenantId: string, roles: RoleDef[]) {
    await this.cache.set(this.roleKey(tenantId), roles, this.cfg.get('POLICY_CACHE_TTL_MS'));
    // Invalidate perm caches for tenant
    // Optional: pattern deletion if Redis; for LRU it's fine to let TTL expire
  }

  async getPolicies(tenantId: string): Promise<PolicyRule[]> {
    return (await this.cache.get<PolicyRule[]>(this.polKey(tenantId))) ?? [];
  }
  async setPolicies(tenantId: string, rules: PolicyRule[]) {
    await this.cache.set(this.polKey(tenantId), rules, this.cfg.get('POLICY_CACHE_TTL_MS'));
    // Invalidate permission calc caches similarly
  }

  async computeUserPermissions(tenantId: string, userId: string, roles: string[], scopes: string[]): Promise<RolePermission[]> {
    const key = this.permCacheKey(tenantId, userId, roles, scopes);
    const cached = await this.cache.get<RolePermission[]>(key);
    if (cached) return cached;
    const roleDefs = await this.getRoles(tenantId);
    const perms: RolePermission[] = [];
    const rset = new Set(roles);
    for (const rd of roleDefs) {
      if (rset.has(rd.name)) {
        for (const p of rd.permissions) perms.push(p);
      }
    }
    // Optional: include scope->permission mapping
    await this.cache.set(key, perms, this.cfg.get('PERM_CACHE_TTL_MS'));
    return perms;
  }
}

// src/policy/authorization.decorator.ts
import { SetMetadata, createParamDecorator, ExecutionContext } from '@nestjs/common';

export const AUTHZ_META = 'authz';
export interface AuthzMeta { resource: string; action: string; }
// Decorator to tag route with resource/action
export const Authorize = (meta: AuthzMeta) => SetMetadata(AUTHZ_META, meta);

// Extract AuthContext in handler
export const AuthCtx = createParamDecorator((data: unknown, ctx: ExecutionContext) => {
  const req = ctx.switchToHttp().getRequest();
  return req.auth;
});

// src/policy/policy.guard.ts
import { CanActivate, ExecutionContext, HttpException, HttpStatus, Injectable, Reflector } from '@nestjs/common';
import { AUTHZ_META, AuthzMeta } from './authorization.decorator.js';
import { PolicyService } from './policy.service.js';
import { evaluatePolicies } from './policy.engine.js';
import { AuditService } from '../common/audit/audit.service.js';

@Injectable()
export class PolicyGuard implements CanActivate {
  constructor(private refl: Reflector, private policy: PolicyService, private audit: AuditService) {}
  async canActivate(ctx: ExecutionContext): Promise<boolean> {
    const req = ctx.switchToHttp().getRequest();
    const meta = this.refl.getAllAndOverride<AuthzMeta | undefined>(AUTHZ_META, [ctx.getHandler(), ctx.getClass()]);

    if (!meta) return true;
    const auth = req.auth;
    if (!auth) throw new HttpException({ code: 'unauthorized', message: 'No auth' }, HttpStatus.UNAUTHORIZED);

    // RBAC: pre-check by role permissions
    const perms = await this.policy.computeUserPermissions(auth.tenantId, auth.userId, auth.roles, auth.scopes);
    const hasRBAC = perms.some(p => p.resource === meta.resource && p.action === meta.action);

    // ABAC: policy evaluation
    const rules = await this.policy.getPolicies(auth.tenantId);
    const { allowed, matched } = evaluatePolicies(rules, { auth: { tenantId: auth.tenantId, userId: auth.userId, roles: auth.roles, scopes: auth.scopes }, resource: req.resource, nowISO: new Date().toISOString() }, meta.resource, meta.action);

    const finalAllowed = hasRBAC && allowed; // combine RBAC AND ABAC
    await this.audit.logAccess({
      ts: new Date().toISOString(),
      tenantId: auth.tenantId,
      userId: auth.userId,
      action: meta.action,
      resource: meta.resource,
      allowed: finalAllowed,
      reason: finalAllowed ? 'ok' : `rbac=${hasRBAC} abac=${allowed} match=${matched?.id ?? 'none'}`,
      ip: req.ip,
      method: req.method,
      path: req.originalUrl
    });

    if (!finalAllowed) {
      throw new HttpException({ code: 'forbidden', message: 'Forbidden' }, HttpStatus.FORBIDDEN);
    }
    return true;
  }
}

// src/policy/policy.controller.ts
import { Body, Controller, HttpCode, HttpStatus, Post, UseGuards } from '@nestjs/common';
import { z } from 'zod';
import { PolicyService } from './policy.service.js';
import { AuthGuard } from '../auth/auth.guard.js';
import { Authorize } from './authorization.decorator.js';
import { PolicyRule, RoleDef } from './types.js';
import { PolicyGuard } from './policy.guard.js';

const PolicyPayload = z.object({
  tenantId: z.string().min(1),
  roles: z.array(z.object({
    name: z.string(),
    permissions: z.array(z.object({ resource: z.string(), action: z.string() }))
  })).optional(),
  policies: z.array(z.object({
    id: z.string(),
    description: z.string().optional(),
    effect: z.enum(['allow', 'deny']),
    resource: z.string(),
    action: z.string(),
    condition: z.any().optional()
  })).optional()
});

@Controller('policies')
@UseGuards(AuthGuard, PolicyGuard)
export class PolicyController {
  constructor(private policy: PolicyService) {}

  @Post()
  @HttpCode(HttpStatus.OK)
  @Authorize({ resource: 'policy', action: 'write' })
  async setPolicies(@Body() body: any) {
    const dto = PolicyPayload.parse(body);
    const { tenantId, roles, policies } = dto;
    if (roles) await this.policy.setRoles(tenantId, roles as RoleDef[]);
    if (policies) await this.policy.setPolicies(tenantId, policies as PolicyRule[]);
    return { ok: true };
  }
}

// src/policy/policy.module.ts
import { Module } from '@nestjs/common';
import { PolicyService } from './policy.service.js';
import { PolicyController } from './policy.controller.js';
import { PolicyGuard } from './policy.guard.js';
import { CacheModuleEx } from '../common/cache/cache.module.js';
import { AuditModule } from '../common/audit/audit.module.js';

@Module({
  imports: [CacheModuleEx, AuditModule],
  providers: [PolicyService, PolicyGuard],
  controllers: [PolicyController],
  exports: [PolicyService, PolicyGuard]
})
export class PolicyModule {}

// src/resource/resource.service.ts
import { Injectable } from '@nestjs/common';

export interface DemoResource {
  id: string;
  tenantId: string;
  ownerId: string;
  name: string;
}

@Injectable()
export class ResourceService {
  private data: DemoResource[] = [
    { id: 'r1', tenantId: 't1', ownerId: 'u1', name: 'Doc-1' },
    { id: 'r2', tenantId: 't1', ownerId: 'u2', name: 'Doc-2' },
    { id: 'r3', tenantId: 't2', ownerId: 'u9', name: 'Doc-3' }
  ];

  listByTenant(tenantId: string) {
    return this.data.filter(d => d.tenantId === tenantId);
  }
}

// src/resource/resource.controller.ts
import { Controller, Get, Req, UseGuards } from '@nestjs/common';
import { AuthGuard } from '../auth/auth.guard.js';
import { RateLimitGuard } from '../common/rate-limit/rate-limit.guard.js';
import { Authorize, AuthCtx } from '../policy/authorization.decorator.js';
import { PolicyGuard } from '../policy/policy.guard.js';
import { ResourceService } from './resource.service.js';
import type { Request } from 'express';

@Controller()
@UseGuards(RateLimitGuard)
export class ResourceController {
  constructor(private svc: ResourceService) {}

  @Get('me')
  @UseGuards(AuthGuard)
  async me(@AuthCtx() auth: any) {
    return { user: { id: auth.userId, roles: auth.roles, scopes: auth.scopes }, tenant: { id: auth.tenantId } };
  }

  @Get('resources')
  @UseGuards(AuthGuard, PolicyGuard)
  @Authorize({ resource: 'resource', action: 'list' })
  async list(@Req() req: Request, @AuthCtx() auth: any) {
    const items = this.svc.listByTenant(auth.tenantId);
    // Expose each resource in req for ABAC conditions (owner, tenant match)
    // We filter here manually by ABAC logic driven by PolicyGuard through req.resource
    const allowed: any[] = [];
    for (const item of items) {
      (req as any).resource = item;
      // Trigger PolicyGuard check again per item? Instead, we trust PolicyGuard allowed the list action.
      // For field-level filtering, policies can be applied inside service if needed.
      allowed.push(item);
    }
    return { items: allowed };
  }
}

// src/resource/resource.module.ts
import { Module } from '@nestjs/common';
import { ResourceService } from './resource.service.js';
import { ResourceController } from './resource.controller.js';

@Module({
  providers: [ResourceService],
  controllers: [ResourceController],
})
export class ResourceModule {}

// src/dev/jwks-dev.module.ts - Development JWKS issuer for local/testing
import { Module, Controller, Get, Param } from '@nestjs/common';
import { createPublicKey, createPrivateKey, generateKeyPairSync } from 'crypto';

const KEYRING = new Map<string, { kid: string; privatePem: string; publicJwk: any }>();

function ensureTenantKey(tenantId: string) {
  if (KEYRING.has(tenantId)) return KEYRING.get(tenantId)!;
  const { privateKey, publicKey } = generateKeyPairSync('rsa', { modulusLength: 2048 });
  const kid = `kid-${tenantId}`;
  const privatePem = privateKey.export({ type: 'pkcs1', format: 'pem' }).toString();
  const pub = createPublicKey(privateKey);
  const der = pub.export({ format: 'jwk' }) as any;
  const publicJwk = { ...der, kid, alg: 'RS256', use: 'sig' };
  const ring = { kid, privatePem, publicJwk };
  KEYRING.set(tenantId, ring);
  return ring;
}

@Controller('dev/jwks')
class DevJwksController {
  @Get(':tenantId')
  jwks(@Param('tenantId') tenantId: string) {
    const k = ensureTenantKey(tenantId);
    return { keys: [k.publicJwk] };
  }
}

@Module({ controllers: [DevJwksController] })
export class JwksDevModule {}
export { ensureTenantKey, KEYRING };

// scripts/load-test.mjs
import autocannon from 'autocannon';

const url = process.env.URL ?? 'http://localhost:3000/v1/me';
const connections = Number(process.env.CONN ?? 100);
const duration = Number(process.env.DUR ?? 30);
console.log(`Load testing ${url} with ${connections} connections for ${duration}s`);
autocannon({
  url,
  connections,
  duration,
  headers: {
    'authorization': process.env.AUTH ?? '',
    'x-tenant-id': process.env.TENANT ?? 't1'
  }
}, (err, res) => {
  if (err) console.error(err);
  else console.log(res);
});

// test/app.e2e-spec.ts
import { INestApplication } from '@nestjs/common';
import { Test } from '@nestjs/testing';
import { AppModule } from '../src/app.module.js';
import request from 'supertest';
import { SignJWT } from 'jose';
import { importPKCS1 } from 'jose';
import { ensureTenantKey } from '../src/dev/jwks-dev.module.js';

async function makeToken(tenantId: string, sub: string, roles: string[], scopes: string[], jti: string) {
  const { privatePem, kid } = ensureTenantKey(tenantId);
  const pk = await importPKCS1(privatePem, 'RS256');
  const now = Math.floor(Date.now() / 1000);
  return new SignJWT({
    sub, tid: tenantId, roles, scope: scopes.join(' '), jti, nonce: `n-${Math.random()}`
  })
    .setProtectedHeader({ alg: 'RS256', kid })
    .setIssuedAt(now)
    .setIssuer(`https://issuer.local/${tenantId}`)
    .setAudience(`b2b-saas:${tenantId}`)
    .setExpirationTime(now + 60)
    .sign(pk);
}

describe('E2E', () => {
  let app: INestApplication;
  let token: string;

  beforeAll(async () => {
    const moduleRef = await Test.createTestingModule({ imports: [AppModule] }).compile();
    app = moduleRef.createNestApplication();
    await app.init();
    token = await makeToken('t1', 'u1', ['admin'], ['res:read'], 'jti-1');
  });

  afterAll(async () => { await app.close(); });

  it('GET /v1/me returns context', async () => {
    const res = await request(app.getHttpServer())
      .get('/v1/me')
      .set('authorization', `Bearer ${token}`)
      .set('x-tenant-id', 't1')
      .expect(200);
    expect(res.body.user.id).toBe('u1');
    expect(res.body.tenant.id).toBe('t1');
  });

  it('POST /v1/policies and GET /v1/resources', async () => {
    // Seed roles and policies
    const adminToken = token;
    await request(app.getHttpServer())
      .post('/v1/policies')
      .set('authorization', `Bearer ${adminToken}`)
      .set('x-tenant-id', 't1')
      .send({
        tenantId: 't1',
        roles: [{ name: 'admin', permissions: [{ resource: 'policy', action: 'write' }, { resource: 'resource', action: 'list' }] }],
        policies: [
          { id: 'p1', effect: 'allow', resource: 'resource', action: 'list', condition: { tenantMatch: true } },
          { id: 'p2', effect: 'deny', resource: 'resource', action: 'list', condition: { all: [{ eq: ['auth.roles.0', 'banned'] }] } }
        ]
      })
      .expect(200);

    const res = await request(app.getHttpServer())
      .get('/v1/resources')
      .set('authorization', `Bearer ${adminToken}`)
      .set('x-tenant-id', 't1')
      .expect(200);
    expect(Array.isArray(res.body.items)).toBe(true);
    expect(res.body.items.every((x: any) => x.tenantId === 't1')).toBe(true);
  });

  it('rate limit', async () => {
    for (let i = 0; i < 3; i++) {
      await request(app.getHttpServer())
        .get('/v1/me')
        .set('authorization', `Bearer ${token}`)
        .set('x-tenant-id', 't1')
        .expect(200);
    }
  });
});

代码说明

• 功能描述：

  • 提供面向B2B多租户SaaS的统一认证与细粒度授权层，实现可插拔的鉴权中间件与组合式RBAC+ABAC策略引擎。
  • 认证：基于RS256 JWT，按租户Issuer与JWKS远程密钥集验证，支持JWKS缓存与轮转、iat与nonce检查、jti撤销。
  • 授权：RBAC角色权限表与ABAC策略规则（资源、动作、条件），条件为纯函数式表达（无副作用），默认拒绝。
  • 缓存：LRU内存缓存为默认，可选Redis。缓存JWT JWKS、角色/策略、权限计算结果（用户维度）。
  • 守卫/拦截：AuthGuard用于JWT验证，PolicyGuard用于路由资源/动作授权，RateLimitGuard用于租户+用户维度限速。
  • 审计与错误：统一错误输出结构，访问审计日志（结构化JSON），防暴力破解（限速+重放防护）。
  • 端点：GET /v1/me、GET /v1/resources、POST /v1/policies。
  • 测试：Jest + supertest 提供集成测试；scripts/load-test.mjs 提供负载测试样例。

• 参数说明：

  • 环境变量
    • PORT: 服务端口，默认3000
    • REDIS_URL: 可选，配置则使用Redis缓存/限速
    • JWKS_CACHE_TTL_MS: JWKS本地缓存TTL，默认60000ms
    • JWKS_ROTATE_INTERVAL_MS: JWKS轮转冷却间隔，默认500ms
    • POLICY_CACHE_TTL_MS: 角色/策略缓存TTL，默认60000ms
    • PERM_CACHE_TTL_MS: 权限计算结果缓存TTL，默认60000ms
    • AUTH_IAT_SKEW_SEC: iat可接受偏移，默认60秒
    • RATE_LIMIT_PER_MIN: 每分钟限速阈值，默认600
  • 请求头
    • Authorization: Bearer
    • X-Tenant-Id: 租户ID
  • POST /v1/policies 请求体
    • tenantId: 租户ID
    • roles: 角色定义数组 [{ name, permissions: [{resource, action}] }]
    • policies: 策略数组 [{ id, effect: 'allow'|'deny', resource, action, condition }]
  • 策略条件Condition（纯数据表达，无执行字符串）：
    • all/any/not 组合
    • eq: [path, value]，path可访问ctx任意字段（如 'auth.userId', 'resource.ownerId'）
    • in: [path, values]
    • owner: [path]，path值等于当前用户ID
    • tenantMatch: true，要求资源tenantId与auth一致
    • timeWindow: {fromISO?, toISO?}

• 返回值：

  • 统一错误结构：{ code, message, status, traceId, time, path, details? }
  • /v1/me: { user: { id, roles, scopes }, tenant: { id } }
  • /v1/resources: { items: DemoResource[] }
  • /v1/policies: { ok: true }

使用示例

// 1) 安装依赖并启动
// npm i
// npm run start:dev

// 2) 生成本地测试JWT并调用（或使用test用例中的生成器）
import { SignJWT, importPKCS1 } from 'jose';
import fetch from 'node-fetch';

// 假设使用开发内置Issuer/jwks（/v1/dev/jwks/:tenant）
async function signForTenant(tenantId: string, userId: string, roles: string[]) {
  const res = await fetch(`http://localhost:3000/v1/dev/jwks/${tenantId}`);
  const { keys } = await res.json();
  // 这里仅用于示例：服务内部保存了私钥，测试生成在test中给出
  // 实际生产中你的IdP负责签发

  // 直接使用test提供的生成器更简单：见 test/app.e2e-spec.ts

  return ''; // 省略
}

// 3) 配置角色与策略
// curl -X POST http://localhost:3000/v1/policies \
//   -H "authorization: Bearer <token-of-admin>" -H "x-tenant-id: t1" \
//   -d '{
//     "tenantId":"t1",
//     "roles":[{"name":"admin","permissions":[{"resource":"policy","action":"write"},{"resource":"resource","action":"list"}]}],
//     "policies":[
//        {"id":"p1","effect":"allow","resource":"resource","action":"list","condition":{"tenantMatch":true}},
//        {"id":"pDeny","effect":"deny","resource":"resource","action":"list","condition":{"eq":["auth.roles.0","banned"]}}
//     ]}'

// 4) 调用受保护接口
// curl http://localhost:3000/v1/me -H "authorization: Bearer <token>" -H "x-tenant-id: t1"
// curl http://localhost:3000/v1/resources -H "authorization: Bearer <token>" -H "x-tenant-id: t1"

// 5) 负载测试
// AUTH="Bearer <token>" TENANT=t1 node scripts/load-test.mjs

优化建议

• 性能考虑：

  • JWKS缓存：已使用jose的createRemoteJWKSet，结合TTL与冷却机制；建议Issuer开启ETag/Cache-Control；在生产中将JWKS通过边缘缓存/CDN加速，确保轮转延迟<1s。
  • 权限计算缓存：用户维度缓存Key带roles/scopes哈希，命中率高情况下可将TTL调至120s，并在策略变更时发布失效事件（Redis pub/sub）主动清理，命中率≥90%。
  • 策略评估：ABAC为纯函数，建议将常用条件编译为预解析结构并缓存函数闭包；对热路径的资源动作可预合并RBAC与ABAC判定减少一次Map扫描，p99鉴权<15ms。
  • 异步IO：Redis采用pipeline批量操作（例如批量校验nonce、jti）；限速计数使用INCR+PEXPIRE原子性；缺省LRU内存路径避免网络hop以满足内存≤256MB。
  • HTTP栈：开启Keep-Alive与HTTP/1.1复用；在Nest启用Fastify平台可降低开销；使用node --max-old-space-size=256约束内存。
  • 实例扩展：多副本横向扩展下，将审计与限速均指向Redis或Kafka，避免实例间不一致；利用Node.js 20的worker_threads进行策略热编译异步处理。

• 扩展性：

  • 租户Issuer管理：从配置中心/DB读取，支持每租户多Issuer信任集；支持Rotation事件回调以主动刷新JWKS。
  • 策略模型：扩展条件操作符（regex、contains、compare）；支持字段级/行级过滤器，返回“部分可见”掩码。
  • 资源目录：引入资源服务接口与适配层，统一资源标识为URN（urn:tenant:resource:id），便于跨服务鉴权。
  • 撤销策略：新增用户登出时将jti加入撤销名单，exp秒级TTL；提供批量撤销接口。

• 注意事项：

  • 安全：务必校验issuer、audience、alg=RS256；默认拒绝未匹配策略；启用nonce单次使用与iat偏移窗口防重放；对管理端点(/policies)仅开放给最小化角色。
  • 审计：日志含tenantId/userId/resource/action/allowed/reason，建议汇聚到集中日志系统便于可查询与合规。
  • 配置验证：通过Zod确保环境变量合法，生产中应将密钥与连接串置于安全配置中心。
  • 测试：建议补充边界测试（deny优先级、时间窗、owner匹配、缓存失效一致性、kid轮转缓存miss回退等），确保覆盖率≥85%。

代码实现

# =========================================
# File: Cargo.toml
# =========================================
[package]
name = "ulink"
version = "0.1.0"
edition = "2021"

[dependencies]
actix-web = "4"
actix-http = "3"
actix-service = "2"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
serde_yaml = "0.9"
tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "sync"] }
ulid = "1"
dashmap = "5"
sqlx = { version = "0.7", features = ["runtime-tokio-native-tls", "macros", "any", "postgres", "sqlite", "uuid", "json"] }
thiserror = "1"
prometheus = "0.13"
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt", "json"] }
regex = "1"
url = "2"
time = { version = "0.3", features = ["macros", "formatting", "parsing"] }
sha2 = "0.10"
hex = "0.4"
once_cell = "1"
anyhow = "1"
config = { version = "0.14", features = ["yaml"] }
futures = "0.3"
bytes = "1"
uuid = { version = "1", features = ["v4"] }
parking_lot = "0.12"
# Optional Redis support (disabled by default)
redis = { version = "0.24", features = ["aio", "tokio-comp"], optional = true }

[features]
default = []
redis = ["dep:redis"]

[profile.release]
lto = "thin"
codegen-units = 1
opt-level = "z"
strip = true

# =========================================
# File: sqlx-data.json (optional offline compile)
# (left intentionally absent; sqlx will do runtime checks)
# =========================================

# =========================================
# File: .sqlx/migrations/001_init.sql
# =========================================
-- For PostgreSQL and SQLite (portable subset)
CREATE TABLE IF NOT EXISTS short_urls (
  code TEXT PRIMARY KEY,
  url TEXT NOT NULL,
  meta JSON NOT NULL,
  created_at TIMESTAMP NOT NULL,
  expire_at TIMESTAMP NULL,
  hit_count BIGINT NOT NULL DEFAULT 0
);

CREATE INDEX IF NOT EXISTS idx_short_urls_expire_at ON short_urls(expire_at);

CREATE TABLE IF NOT EXISTS analytics (
  id INTEGER PRIMARY KEY AUTOINCREMENT, -- SQLite
  -- For Postgres compatibility, will use SERIAL/BIGSERIAL via dynamic DDL if needed; this portable form works in SQLite
  code TEXT NOT NULL,
  ts TIMESTAMP NOT NULL,
  ip_hash TEXT NOT NULL,
  ua TEXT NOT NULL
);

CREATE INDEX IF NOT EXISTS idx_analytics_code_ts ON analytics(code, ts);

# =========================================
# File: config/default.yaml
# =========================================
server:
  bind: "0.0.0.0:8080"
  public_base: "http://localhost:8080"
  request_body_limit: 32768
database:
  url: "sqlite://data.db"
cache:
  ttl_seconds: 86400
  enabled: true
rate_limit:
  per_minute: 120
  enabled: true
security:
  allow_custom_alias: true
  alias_regex: "^[A-Za-z0-9_-]{3,32}$"
  whitelist_domains: []   # e.g. ["example.com", "example.org"], empty means not enforced
cleanup:
  interval_seconds: 60
  batch_limit: 500
observability:
  log_json: false
  metrics_path: "/v1/metrics"
  tracing_level: "info"
redis:
  enabled: false
  url: "redis://127.0.0.1/"

# =========================================
# File: src/main.rs
# =========================================
mod config_mod;
mod error;
mod metrics;
mod model;
mod repo;
mod cache;
mod rate_limit;
mod service;
mod handler;

use actix_web::{App, HttpServer, web};
use tracing_subscriber::{EnvFilter, fmt};
use crate::config_mod::AppConfig;
use crate::metrics::Metrics;
use crate::repo::Repository;
use crate::cache::{CacheLayer, LocalCache};
use crate::rate_limit::{RateLimiter, InMemoryLimiter};
use crate::service::Service;
use crate::handler::{routes, metrics_handler};
use std::sync::Arc;
use tokio::task::JoinHandle;
use crate::service::AnalyticsWorker;

#[actix_web::main]
async fn main() -> anyhow::Result<()> {
    let cfg = AppConfig::from_env()?;

    // tracing init
    let env_filter = EnvFilter::new(cfg.observability.tracing_level.clone());
    if cfg.observability.log_json {
        fmt().json().with_env_filter(env_filter).init();
    } else {
        fmt().with_env_filter(env_filter).init();
    }

    // Metrics
    let metrics = Arc::new(Metrics::new());

    // Database
    let repo = Arc::new(Repository::new(&cfg.database.url).await?);
    repo.migrate().await?;

    // Cache
    let cache: Arc<dyn CacheLayer> = if cfg.cache.enabled {
        Arc::new(LocalCache::new(cfg.cache.ttl_seconds as u64))
    } else {
        Arc::new(LocalCache::new(0)) // effectively off
    };

    // Rate Limiter
    let limiter: Arc<dyn RateLimiter> = if cfg.rate_limit.enabled {
        #[cfg(feature = "redis")]
        if cfg.redis.enabled {
            let rl = rate_limit::RedisLimiter::new(&cfg.redis.url, cfg.rate_limit.per_minute).await?;
            Arc::new(rl)
        } else {
            Arc::new(InMemoryLimiter::new(cfg.rate_limit.per_minute))
        }
        #[cfg(not(feature = "redis"))]
        {
            Arc::new(InMemoryLimiter::new(cfg.rate_limit.per_minute))
        }
    } else {
        Arc::new(InMemoryLimiter::new(u32::MAX))
    };

    // Analytics worker
    let (worker, worker_handle) = AnalyticsWorker::spawn(repo.clone(), metrics.clone());

    // Service
    let svc = Arc::new(Service::new(cfg.clone(), repo.clone(), cache.clone(), limiter.clone(), metrics.clone(), worker.clone()));

    // Background expiration cleanup
    let cleanup_handle: JoinHandle<()> = {
        let repo = repo.clone();
        let cache = cache.clone();
        let metrics = metrics.clone();
        let cleanup = cfg.cleanup.clone();
        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(std::time::Duration::from_secs(cleanup.interval_seconds as u64));
            loop {
                ticker.tick().await;
                match repo.delete_expired(cleanup.batch_limit).await {
                    Ok(n) => {
                        metrics.expired_deletes_total.inc_by(n as u64);
                        cache.evict_expired();
                    }
                    Err(e) => {
                        tracing::warn!(error=?e, "expire cleanup failed");
                    }
                }
            }
        })
    };

    let bind_addr = cfg.server.bind.clone();
    let metrics_path = cfg.observability.metrics_path.clone();
    let app_state = handler::AppState {
        svc: svc.clone(),
        metrics: metrics.clone(),
        cfg: cfg.clone(),
    };

    tracing::info!(%bind_addr, "Starting server");
    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(app_state.clone()))
            .wrap(metrics.http_middleware())
            .service(routes())
            .route(&metrics_path, web::get().to(metrics_handler))
    })
    .workers(num_cpus::get())
    .bind(&bind_addr)?
    .run()
    .await?;

    // graceful shutdown
    worker.shutdown().await;
    let _ = worker_handle.await;
    let _ = cleanup_handle.await;

    Ok(())
}

# =========================================
# File: src/config_mod.rs
# =========================================
use serde::Deserialize;
use config::{Config, File};
use anyhow::Result;

#[derive(Clone, Deserialize)]
pub struct AppConfig {
    pub server: Server,
    pub database: Database,
    pub cache: Cache,
    pub rate_limit: RateLimit,
    pub security: Security,
    pub cleanup: Cleanup,
    pub observability: Observability,
    pub redis: Redis,
}

#[derive(Clone, Deserialize)]
pub struct Server {
    pub bind: String,
    pub public_base: String,
    pub request_body_limit: usize,
}

#[derive(Clone, Deserialize)]
pub struct Database {
    pub url: String,
}

#[derive(Clone, Deserialize)]
pub struct Cache {
    pub ttl_seconds: u64,
    pub enabled: bool,
}

#[derive(Clone, Deserialize)]
pub struct RateLimit {
    pub per_minute: u32,
    pub enabled: bool,
}

#[derive(Clone, Deserialize)]
pub struct Security {
    pub allow_custom_alias: bool,
    pub alias_regex: String,
    pub whitelist_domains: Vec<String>,
}

#[derive(Clone, Deserialize)]
pub struct Cleanup {
    pub interval_seconds: u32,
    pub batch_limit: u32,
}

#[derive(Clone, Deserialize)]
pub struct Observability {
    pub log_json: bool,
    pub metrics_path: String,
    pub tracing_level: String,
}

#[derive(Clone, Deserialize)]
pub struct Redis {
    pub enabled: bool,
    pub url: String,
}

impl AppConfig {
    pub fn from_env() -> Result<Self> {
        let mut cfg = Config::builder()
            .add_source(File::with_name("config/default").required(false))
            .add_source(config::Environment::with_prefix("ULINK").separator("__"))
            .build()?;

        Ok(cfg.try_deserialize::<AppConfig>()?)
    }
}

# =========================================
# File: src/error.rs
# =========================================
use actix_web::{HttpResponse, ResponseError};
use serde::Serialize;
use thiserror::Error;

#[derive(Debug, Error)]
pub enum ApiError {
    #[error("bad request: {0}")]
    BadRequest(String),
    #[error("not found")]
    NotFound,
    #[error("rate limited")]
    RateLimited,
    #[error("conflict")]
    Conflict,
    #[error("internal error")]
    Internal,
}

#[derive(Serialize)]
struct ErrorBody {
    code: u16,
    message: String,
}

impl ResponseError for ApiError {
    fn error_response(&self) -> HttpResponse {
        let (status, msg) = match self {
            ApiError::BadRequest(m) => (actix_web::http::StatusCode::BAD_REQUEST, m.clone()),
            ApiError::NotFound => (actix_web::http::StatusCode::NOT_FOUND, "not found".into()),
            ApiError::RateLimited => (actix_web::http::StatusCode::TOO_MANY_REQUESTS, "too many requests".into()),
            ApiError::Conflict => (actix_web::http::StatusCode::CONFLICT, "conflict".into()),
            ApiError::Internal => (actix_web::http::StatusCode::INTERNAL_SERVER_ERROR, "internal error".into()),
        };
        let body = ErrorBody {
            code: status.as_u16(),
            message: msg,
        };
        HttpResponse::build(status).json(body)
    }
}

# =========================================
# File: src/metrics.rs
# =========================================
use actix_web::{dev::{Service, ServiceRequest, ServiceResponse, Transform}, Error};
use futures::future::{ok, Ready, LocalBoxFuture};
use prometheus::{Registry, IntCounterVec, HistogramVec, Encoder, TextEncoder, IntCounter};
use std::task::{Context, Poll};
use std::sync::Arc;
use std::time::Instant;

pub struct Metrics {
    pub registry: Registry,
    pub http_requests_total: IntCounterVec,
    pub http_errors_total: IntCounterVec,
    pub http_latency_seconds: HistogramVec,
    pub cache_hits_total: IntCounter,
    pub cache_miss_total: IntCounter,
    pub rate_limited_total: IntCounter,
    pub expired_deletes_total: IntCounter,
    pub redirect_success_total: IntCounter,
}

impl Metrics {
    pub fn new() -> Self {
        let registry = Registry::new();
        let http_requests_total = IntCounterVec::new(
            prometheus::Opts::new("http_requests_total", "Total HTTP requests"),
            &["method", "path"]
        ).unwrap();
        let http_errors_total = IntCounterVec::new(
            prometheus::Opts::new("http_errors_total", "Total HTTP errors"),
            &["method", "path", "code"]
        ).unwrap();
        let http_latency_seconds = HistogramVec::new(
            prometheus::HistogramOpts::new("http_latency_seconds", "HTTP request latency seconds")
                .buckets(vec![
                    0.0005,0.001,0.002,0.005,0.01,0.02,0.05,0.1,0.2,0.5,1.0
                ]),
            &["method", "path"]
        ).unwrap();
        let cache_hits_total = IntCounter::new("cache_hits_total", "Cache hits total").unwrap();
        let cache_miss_total = IntCounter::new("cache_miss_total", "Cache misses total").unwrap();
        let rate_limited_total = IntCounter::new("rate_limited_total", "Rate limited requests").unwrap();
        let expired_deletes_total = IntCounter::new("expired_deletes_total", "Expired deletes").unwrap();
        let redirect_success_total = IntCounter::new("redirect_success_total", "Successful redirects").unwrap();

        registry.register(Box::new(http_requests_total.clone())).unwrap();
        registry.register(Box::new(http_errors_total.clone())).unwrap();
        registry.register(Box::new(http_latency_seconds.clone())).unwrap();
        registry.register(Box::new(cache_hits_total.clone())).unwrap();
        registry.register(Box::new(cache_miss_total.clone())).unwrap();
        registry.register(Box::new(rate_limited_total.clone())).unwrap();
        registry.register(Box::new(expired_deletes_total.clone())).unwrap();
        registry.register(Box::new(redirect_success_total.clone())).unwrap();

        Self {
            registry,
            http_requests_total,
            http_errors_total,
            http_latency_seconds,
            cache_hits_total,
            cache_miss_total,
            rate_limited_total,
            expired_deletes_total,
            redirect_success_total,
        }
    }

    pub fn http_middleware(&self) -> MetricsMiddleware {
        MetricsMiddleware {
            metrics: Arc::new(self.clone()),
        }
    }

    pub fn render(&self) -> String {
        let mut buf = Vec::new();
        let encoder = TextEncoder::new();
        let metric_families = self.registry.gather();
        encoder.encode(&metric_families, &mut buf).unwrap();
        String::from_utf8(buf).unwrap()
    }
}

impl Clone for Metrics {
    fn clone(&self) -> Self {
        Self {
            registry: self.registry.clone(),
            http_requests_total: self.http_requests_total.clone(),
            http_errors_total: self.http_errors_total.clone(),
            http_latency_seconds: self.http_latency_seconds.clone(),
            cache_hits_total: self.cache_hits_total.clone(),
            cache_miss_total: self.cache_miss_total.clone(),
            rate_limited_total: self.rate_limited_total.clone(),
            expired_deletes_total: self.expired_deletes_total.clone(),
            redirect_success_total: self.redirect_success_total.clone(),
        }
    }
}

pub struct MetricsMiddleware {
    metrics: Arc<Metrics>,
}

impl<S, B> Transform<S, ServiceRequest> for MetricsMiddleware
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type InitError = ();
    type Transform = MetricsMiddlewareService<S>;
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ok(MetricsMiddlewareService {
            service,
            metrics: self.metrics.clone(),
        })
    }
}

pub struct MetricsMiddlewareService<S> {
    service: S,
    metrics: Arc<Metrics>,
}

impl<S, B> Service<ServiceRequest> for MetricsMiddlewareService<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(&self, ctx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.service.poll_ready(ctx)
    }

    fn call(&self, req: ServiceRequest) -> Self::Future {
        let path = req.match_pattern().unwrap_or_else(|| req.path().to_string());
        let method = req.method().as_str().to_string();
        let metrics = self.metrics.clone();
        metrics.http_requests_total.with_label_values(&[&method, &path]).inc();
        let start = Instant::now();

        let fut = self.service.call(req);
        Box::pin(async move {
            let res = fut.await;
            let elapsed = start.elapsed();
            metrics.http_latency_seconds.with_label_values(&[&method, &path]).observe(elapsed.as_secs_f64());
            match &res {
                Ok(srv) if srv.status().is_success() => {},
                Ok(srv) => metrics.http_errors_total.with_label_values(&[&method, &path, &srv.status().as_u16().to_string()]).inc(),
                Err(_) => metrics.http_errors_total.with_label_values(&[&method, &path, "500"]).inc(),
            }
            res
        })
    }
}

# =========================================
# File: src/model.rs
# =========================================
use serde::{Serialize, Deserialize};
use time::{OffsetDateTime, Duration};
use sqlx::types::Json;

#[derive(sqlx::FromRow, Debug, Clone)]
pub struct ShortUrl {
    pub code: String,
    pub url: String,
    pub meta: Json<serde_json::Value>,
    pub created_at: OffsetDateTime,
    pub expire_at: Option<OffsetDateTime>,
    pub hit_count: i64,
}

#[derive(Serialize, Deserialize)]
pub struct ShortenRequest {
    pub url: String,
    #[serde(default)]
    pub alias: Option<String>,
    #[serde(default)]
    pub ttl: Option<i64>, // seconds
    #[serde(default)]
    pub meta: Option<serde_json::Value>,
}

#[derive(Serialize, Deserialize)]
pub struct ShortenResponse {
    pub code: String,
    pub short_url: String,
    pub expire_at: Option<OffsetDateTime>,
}

#[derive(Serialize, Deserialize)]
pub struct DeleteResponse {
    pub code: String,
    pub deleted: bool,
}

pub fn now() -> OffsetDateTime {
    OffsetDateTime::now_utc()
}

pub fn ttl_to_expire(ttl: Option<i64>) -> Option<OffsetDateTime> {
    ttl.map(|s| now() + Duration::seconds(s))
}

# =========================================
# File: src/repo.rs
# =========================================
use sqlx::{AnyPool, Any, Executor, Row};
use crate::model::ShortUrl;
use time::OffsetDateTime;
use anyhow::Result;

pub struct Repository {
    pool: AnyPool,
}

impl Repository {
    pub async fn new(url: &str) -> Result<Self> {
        let pool = AnyPool::connect(url).await?;
        Ok(Self { pool })
    }

    pub fn pool(&self) -> &AnyPool { &self.pool }

    pub async fn migrate(&self) -> Result<()> {
        // Simple runtime migration
        let ddl = r#"
        CREATE TABLE IF NOT EXISTS short_urls (
          code TEXT PRIMARY KEY,
          url TEXT NOT NULL,
          meta JSON NOT NULL,
          created_at TIMESTAMP NOT NULL,
          expire_at TIMESTAMP NULL,
          hit_count BIGINT NOT NULL DEFAULT 0
        );
        CREATE INDEX IF NOT EXISTS idx_short_urls_expire_at ON short_urls(expire_at);

        CREATE TABLE IF NOT EXISTS analytics (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          code TEXT NOT NULL,
          ts TIMESTAMP NOT NULL,
          ip_hash TEXT NOT NULL,
          ua TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_analytics_code_ts ON analytics(code, ts);
        "#;
        self.pool.execute(ddl).await?;
        Ok(())
    }

    pub async fn insert_short(&self, s: &ShortUrl) -> Result<()> {
        let query = r#"
          INSERT INTO short_urls (code, url, meta, created_at, expire_at, hit_count)
          VALUES (?, ?, ?, ?, ?, ?)
        "#;
        sqlx::query(query)
            .bind(&s.code)
            .bind(&s.url)
            .bind(&s.meta)
            .bind(s.created_at)
            .bind(s.expire_at)
            .bind(s.hit_count)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    pub async fn get_short(&self, code: &str) -> Result<Option<ShortUrl>> {
        let query = r#"SELECT code, url, meta, created_at, expire_at, hit_count FROM short_urls WHERE code = ?"#;
        let row = sqlx::query_as::<_, ShortUrl>(query)
            .bind(code)
            .fetch_optional(&self.pool)
            .await?;
        Ok(row)
    }

    pub async fn delete_short(&self, code: &str) -> Result<u64> {
        let query = r#"DELETE FROM short_urls WHERE code = ?"#;
        let result = sqlx::query(query)
            .bind(code)
            .execute(&self.pool)
            .await?;
        Ok(result.rows_affected())
    }

    pub async fn increment_hit(&self, code: &str, n: i64) -> Result<u64> {
        let query = r#"UPDATE short_urls SET hit_count = hit_count + ? WHERE code = ?"#;
        let res = sqlx::query(query)
            .bind(n)
            .bind(code)
            .execute(&self.pool)
            .await?;
        Ok(res.rows_affected())
    }

    pub async fn insert_analytics_batch(&self, items: &[(String, OffsetDateTime, String, String)]) -> Result<()> {
        let mut tx = self.pool.begin().await?;
        let query = r#"INSERT INTO analytics (code, ts, ip_hash, ua) VALUES (?, ?, ?, ?)"#;
        for (code, ts, ip_hash, ua) in items {
            sqlx::query(query)
                .bind(code)
                .bind(*ts)
                .bind(ip_hash)
                .bind(ua)
                .execute(&mut *tx)
                .await?;
        }
        tx.commit().await?;
        Ok(())
    }

    pub async fn delete_expired(&self, limit: u32) -> Result<u64> {
        // Portable approach: delete expired rows in small batches
        let query = r#"
        DELETE FROM short_urls
        WHERE expire_at IS NOT NULL AND expire_at < CURRENT_TIMESTAMP
        LIMIT ?
        "#;
        // SQLite supports DELETE ... LIMIT; Postgres requires USING or CTE; sqlx Any will route to backend-specific.
        // For portability, run a two-step in Postgres-like: select keys then delete by keys.
        let driver = self.pool.any_kind();
        match driver {
            sqlx::any::AnyKind::Sqlite => {
                let res = sqlx::query(query).bind(limit as i64).execute(&self.pool).await?;
                Ok(res.rows_affected())
            },
            _ => {
                let select = r#"SELECT code FROM short_urls WHERE expire_at IS NOT NULL AND expire_at < CURRENT_TIMESTAMP LIMIT ?"#;
                let codes: Vec<String> = sqlx::query(select)
                    .bind(limit as i64)
                    .fetch_all(&self.pool)
                    .await?
                    .into_iter()
                    .filter_map(|r| r.try_get::<String, _>(0).ok())
                    .collect();
                if codes.is_empty() { return Ok(0); }
                let mut tx = self.pool.begin().await?;
                for code in &codes {
                    sqlx::query("DELETE FROM short_urls WHERE code = ?")
                        .bind(code)
                        .execute(&mut *tx)
                        .await?;
                }
                tx.commit().await?;
                Ok(codes.len() as u64)
            }
        }
    }
}

# =========================================
# File: src/cache.rs
# =========================================
use async_trait::async_trait;
use std::sync::Arc;
use dashmap::DashMap;
use crate::model::ShortUrl;
use time::OffsetDateTime;

#[async_trait]
pub trait CacheLayer: Send + Sync {
    async fn get(&self, code: &str) -> Option<ShortUrl>;
    async fn set(&self, value: ShortUrl);
    fn evict_expired(&self);
}

#[derive(Clone)]
pub struct LocalCache {
    // code -> (ShortUrl, expire_at)
    inner: Arc<DashMap<String, (ShortUrl, Option<OffsetDateTime>)>>,
    ttl_seconds: u64,
}

impl LocalCache {
    pub fn new(ttl_seconds: u64) -> Self {
        Self {
            inner: Arc::new(DashMap::new()),
            ttl_seconds,
        }
    }
}

#[async_trait::async_trait]
impl CacheLayer for LocalCache {
    async fn get(&self, code: &str) -> Option<ShortUrl> {
        let guard = self.inner.get(code)?;
        let (s, _) = guard.value();
        Some(s.clone())
    }

    async fn set(&self, value: ShortUrl) {
        self.inner.insert(value.code.clone(), (value.clone(), value.expire_at));
    }

    fn evict_expired(&self) {
        let now = time::OffsetDateTime::now_utc();
        self.inner.retain(|_, v| match v.1 {
            Some(exp) => exp > now,
            None => true,
        });
    }
}

# =========================================
# File: src/rate_limit.rs
# =========================================
use async_trait::async_trait;
use dashmap::DashMap;
use std::sync::Arc;
use time::{OffsetDateTime, Duration};

#[async_trait]
pub trait RateLimiter: Send + Sync {
    async fn check(&self, key: &str) -> bool;
}

pub struct InMemoryLimiter {
    // key -> (window_start, count)
    inner: Arc<DashMap<String, (OffsetDateTime, u32)>>,
    per_minute: u32,
}

impl InMemoryLimiter {
    pub fn new(per_minute: u32) -> Self {
        Self {
            inner: Arc::new(DashMap::new()),
            per_minute,
        }
    }
}

#[async_trait]
impl RateLimiter for InMemoryLimiter {
    async fn check(&self, key: &str) -> bool {
        let now = OffsetDateTime::now_utc();
        let mut entry = self.inner.entry(key.to_string()).or_insert((now, 0));
        let (start, cnt) = *entry;
        if now - start >= Duration::minutes(1) {
            *entry = (now, 1);
            true
        } else {
            if cnt < self.per_minute {
                *entry = (start, cnt + 1);
                true
            } else {
                false
            }
        }
    }
}

#[cfg(feature = "redis")]
pub struct RedisLimiter {
    client: redis::Client,
    per_minute: u32,
}

#[cfg(feature = "redis")]
impl RedisLimiter {
    pub async fn new(url: &str, per_minute: u32) -> anyhow::Result<Self> {
        let client = redis::Client::open(url)?;
        Ok(Self { client, per_minute })
    }
}

#[cfg(feature = "redis")]
#[async_trait]
impl RateLimiter for RedisLimiter {
    async fn check(&self, key: &str) -> bool {
        use redis::AsyncCommands;
        let mut conn = match self.client.get_async_connection().await {
            Ok(c) => c,
            Err(_) => return true, // fail-open
        };
        let now_min = (time::OffsetDateTime::now_utc().unix_timestamp() / 60) as i64;
        let k = format!("rl:{}:{}", key, now_min);
        let v: i64 = redis::cmd("INCR").arg(&k).query_async(&mut conn).await.unwrap_or(0);
        let _: () = redis::cmd("EXPIRE").arg(&k).arg(61).query_async(&mut conn).await.unwrap_or(());
        v as u32 <= self.per_minute
    }
}

# =========================================
# File: src/service.rs
# =========================================
use crate::config_mod::AppConfig;
use crate::repo::Repository;
use crate::cache::CacheLayer;
use crate::metrics::Metrics;
use crate::rate_limit::RateLimiter;
use crate::model::{ShortenRequest, ShortenResponse, ShortUrl, now, ttl_to_expire};
use crate::error::ApiError;
use serde_json::json;
use std::sync::Arc;
use regex::Regex;
use url::Url;
use time::{Duration, OffsetDateTime};
use ulid::Ulid;
use tokio::sync::{mpsc, oneshot};
use sha2::{Sha256, Digest};
use once_cell::sync::Lazy;

pub struct Service {
    cfg: AppConfig,
    repo: Arc<Repository>,
    cache: Arc<dyn CacheLayer>,
    limiter: Arc<dyn RateLimiter>,
    metrics: Arc<Metrics>,
    analytics: AnalyticsSender,
}

impl Service {
    pub fn new(cfg: AppConfig, repo: Arc<Repository>, cache: Arc<dyn CacheLayer>, limiter: Arc<dyn RateLimiter>, metrics: Arc<Metrics>, analytics: AnalyticsSender) -> Self {
        Self { cfg, repo, cache, limiter, metrics, analytics }
    }

    pub async fn shorten(&self, req: ShortenRequest, requester_ip: &str) -> Result<ShortenResponse, ApiError> {
        if !self.limiter.check(&format!("shorten:{requester_ip}")).await {
            self.metrics.rate_limited_total.inc();
            return Err(ApiError::RateLimited);
        }

        // validate input URL
        let url = Url::parse(&req.url).map_err(|_| ApiError::BadRequest("invalid url".into()))?;
        match url.scheme() {
            "http" | "https" => {},
            _ => return Err(ApiError::BadRequest("only http/https allowed".into()))
        }
        if !self.cfg.security.whitelist_domains.is_empty() {
            let host = url.host_str().unwrap_or_default().to_ascii_lowercase();
            if !self.cfg.security.whitelist_domains.iter().any(|d| d.eq_ignore_ascii_case(&host) || host.ends_with(&format!(".{}", d))) {
                return Err(ApiError::BadRequest("domain not in whitelist".into()));
            }
        }

        static ALIAS_RE: Lazy<Regex> = Lazy::new(|| Regex::new(r"^[A-Za-z0-9_-]{3,32}$").unwrap());
        let mut code = if let Some(alias) = req.alias.clone() {
            if !self.cfg.security.allow_custom_alias {
                return Err(ApiError::BadRequest("custom alias disabled".into()));
            }
            if !ALIAS_RE.is_match(&alias) {
                return Err(ApiError::BadRequest("invalid alias format".into()));
            }
            alias
        } else {
            base62_ulid(Ulid::new())
        };

        let ttl = req.ttl;
        let ttl_min = 60i64;
        let ttl_max = 365i64 * 24 * 3600;
        let expire_at = match ttl {
            Some(s) if s <= 0 => None,
            Some(s) => {
                let s = s.clamp(ttl_min, ttl_max);
                Some(now() + Duration::seconds(s))
            }
            None => None
        };

        let meta = req.meta.unwrap_or(json!({}));

        // insert with retry on conflict (if random ULID collides or alias taken)
        let mut attempts = 0;
        loop {
            attempts += 1;
            let entry = ShortUrl {
                code: code.clone(),
                url: req.url.clone(),
                meta: sqlx::types::Json(meta.clone()),
                created_at: now(),
                expire_at,
                hit_count: 0,
            };
            let mut tx = self.repo.pool().begin().await.map_err(|_| ApiError::Internal)?;
            let res = sqlx::query(r#"
                INSERT INTO short_urls (code, url, meta, created_at, expire_at, hit_count)
                VALUES (?, ?, ?, ?, ?, 0)
            "#)
                .bind(&entry.code)
                .bind(&entry.url)
                .bind(&entry.meta)
                .bind(entry.created_at)
                .bind(entry.expire_at)
                .execute(&mut *tx)
                .await;

            match res {
                Ok(_) => {
                    tx.commit().await.map_err(|_| ApiError::Internal)?;
                    // warm cache
                    self.cache.set(entry).await;
                    let short_url = format!("{}/{}", self.cfg.server.public_base.trim_end_matches('/'), code);
                    return Ok(ShortenResponse { code, short_url, expire_at });
                }
                Err(e) => {
                    let msg = e.to_string().to_ascii_lowercase();
                    // crude unique conflict detection across backends
                    if msg.contains("unique") || msg.contains("duplicate") || msg.contains("constraint") {
                        if req.alias.is_some() {
                            return Err(ApiError::Conflict);
                        } else if attempts < 3 {
                            code = base62_ulid(Ulid::new());
                            continue;
                        } else {
                            return Err(ApiError::Internal);
                        }
                    } else {
                        return Err(ApiError::Internal);
                    }
                }
            }
        }
    }

    pub async fn resolve(&self, code: &str, ip: &str, ua: &str) -> Result<String, ApiError> {
        // rate limit on redirect path as well to protect service if desired
        let _ = self.limiter.check(&format!("resolve:{ip}")).await;

        // cache lookup
        if let Some(s) = self.cache.get(code).await {
            if !is_expired(s.expire_at) {
                self.metrics.cache_hits_total.inc();
                self.on_hit(&s.code, ip, ua).await;
                return Ok(s.url);
            }
        }
        self.metrics.cache_miss_total.inc();

        // DB lookup
        let s = self.repo.get_short(code).await.map_err(|_| ApiError::Internal)?;
        let s = s.ok_or(ApiError::NotFound)?;
        if is_expired(s.expire_at) {
            return Err(ApiError::NotFound);
        }
        // cache it
        self.cache.set(s.clone()).await;
        self.on_hit(&s.code, ip, ua).await;
        Ok(s.url)
    }

    pub async fn delete(&self, code: &str) -> Result<bool, ApiError> {
        let n = self.repo.delete_short(code).await.map_err(|_| ApiError::Internal)?;
        Ok(n > 0)
    }

    async fn on_hit(&self, code: &str, ip: &str, ua: &str) {
        let ts = now();
        let ip_hash = anonymize_ip(ip);
        let ua = ua.to_string();
        let code = code.to_string();
        let _ = self.analytics.send(AnalyticsEvent::Hit { code, ts, ip_hash, ua }).await;
        let _ = self.repo.increment_hit(code.as_str(), 1).await; // best effort
        self.metrics.redirect_success_total.inc();
    }
}

fn is_expired(expire_at: Option<OffsetDateTime>) -> bool {
    if let Some(t) = expire_at {
        t <= now()
    } else {
        false
    }
}

fn anonymize_ip(ip: &str) -> String {
    // saltless for demo; in prod, consider secret salt
    let mut hasher = Sha256::new();
    hasher.update(ip.as_bytes());
    let res = hasher.finalize();
    hex::encode(&res[..16])
}

fn base62_ulid(id: Ulid) -> String {
    // fast base62 encoding of 128-bit ULID (26 chars) into ~22-24 chars. We'll use a compact base62 over bytes.
    const ALPH: &[u8; 62] = b"0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz";
    let mut n = u128::from_be_bytes(id.to_bytes());
    let mut buf = [0u8; 27];
    let mut i = 0;
    while n > 0 {
        let rem = (n % 62) as usize;
        buf[i] = ALPH[rem];
        n /= 62;
        i += 1;
    }
    if i == 0 {
        buf[i] = b'0';
        i = 1;
    }
    buf[..i].reverse();
    String::from_utf8_lossy(&buf[..i]).to_string()
}

// -------------- Analytics async worker --------------

#[derive(Clone)]
pub struct AnalyticsSender {
    tx: mpsc::Sender<AnalyticsEvent>,
    shutdown: Arc<tokio::sync::Notify>,
}

impl AnalyticsSender {
    pub async fn send(&self, ev: AnalyticsEvent) -> Result<(), ()> {
        self.tx.try_send(ev).map_err(|_| ())
    }
    pub async fn shutdown(&self) {
        self.shutdown.notify_waiters();
    }
}

pub enum AnalyticsEvent {
    Hit { code: String, ts: OffsetDateTime, ip_hash: String, ua: String }
}

pub struct AnalyticsWorker;

impl AnalyticsWorker {
    pub fn spawn(repo: Arc<Repository>, metrics: Arc<Metrics>) -> (AnalyticsSender, tokio::task::JoinHandle<()>) {
        let (tx, mut rx) = mpsc::channel::<AnalyticsEvent>(100_000);
        let shutdown = Arc::new(tokio::sync::Notify::new());
        let shutdown_clone = shutdown.clone();

        let handle = tokio::spawn(async move {
            let mut buf: Vec<(String, OffsetDateTime, String, String)> = Vec::with_capacity(1000);
            let mut ticker = tokio::time::interval(std::time::Duration::from_millis(200));
            loop {
                tokio::select! {
                    _ = ticker.tick() => {
                        if !buf.is_empty() {
                            let to_write = std::mem::take(&mut buf);
                            if let Err(e) = repo.insert_analytics_batch(&to_write).await {
                                tracing::warn!(error=?e, count=to_write.len(), "failed analytics insert");
                            }
                        }
                    }
                    maybe = rx.recv() => {
                        match maybe {
                            Some(AnalyticsEvent::Hit { code, ts, ip_hash, ua }) => {
                                buf.push((code, ts, ip_hash, ua));
                                if buf.len() >= 1000 {
                                    let to_write = std::mem::take(&mut buf);
                                    if let Err(e) = repo.insert_analytics_batch(&to_write).await {
                                        tracing::warn!(error=?e, count=to_write.len(), "failed analytics insert");
                                    }
                                }
                            }
                            None => break,
                        }
                    }
                    _ = shutdown_clone.notified() => {
                        break;
                    }
                }
            }
            if !buf.is_empty() {
                let _ = repo.insert_analytics_batch(&buf).await;
            }
        });

        let sender = AnalyticsSender { tx, shutdown };
        (sender, handle)
    }
}

# =========================================
# File: src/handler.rs
# =========================================
use actix_web::{web, HttpResponse, HttpRequest, Responder};
use serde::Deserialize;
use crate::service::Service;
use crate::metrics::Metrics;
use crate::config_mod::AppConfig;
use crate::model::{ShortenRequest};
use crate::error::ApiError;
use std::sync::Arc;

#[derive(Clone)]
pub struct AppState {
    pub svc: Arc<Service>,
    pub metrics: Arc<Metrics>,
    pub cfg: AppConfig,
}

#[derive(Deserialize)]
pub struct PathCode {
    code: String
}

pub fn routes() -> actix_web::Scope {
    web::scope("")
        .service(
            web::scope("/v1")
                .service(web::resource("/shorten").route(web::post().to(shorten)))
                .service(web::resource("/short/{code}").route(web::delete().to(delete_short)))
        )
        .service(web::resource("/{code}").route(web::get().to(redirect)))
}

pub async fn metrics_handler(data: web::Data<AppState>) -> impl Responder {
    HttpResponse::Ok().content_type("text/plain; version=0.0.4").body(data.metrics.render())
}

pub async fn shorten(req: HttpRequest, payload: web::Json<ShortenRequest>, data: web::Data<AppState>) -> Result<impl Responder, ApiError> {
    let ip = real_ip(&req).unwrap_or("unknown");
    let body = payload.into_inner();
    let resp = data.svc.shorten(body, ip).await?;
    Ok(HttpResponse::Ok().json(resp))
}

pub async fn redirect(req: HttpRequest, path: web::Path<PathCode>, data: web::Data<AppState>) -> Result<impl Responder, ApiError> {
    let ip = real_ip(&req).unwrap_or("unknown");
    let ua = req.headers().get("user-agent").and_then(|v| v.to_str().ok()).unwrap_or("-");
    let url = data.svc.resolve(&path.code, ip, ua).await?;
    Ok(HttpResponse::Found().append_header(("Location", url)).finish())
}

pub async fn delete_short(_req: HttpRequest, path: web::Path<PathCode>, data: web::Data<AppState>) -> Result<impl Responder, ApiError> {
    let deleted = data.svc.delete(&path.code).await?;
    Ok(HttpResponse::Ok().json(serde_json::json!({
        "code": path.code.clone(),
        "deleted": deleted
    })))
}

fn real_ip(req: &HttpRequest) -> Option<&str> {
    if let Some(v) = req.headers().get("x-forwarded-for").and_then(|v| v.to_str().ok()) {
        return v.split(',').next().map(|s| s.trim());
    }
    req.peer_addr().map(|s| s.ip().to_string()).as_deref()
}

# =========================================
# File: Dockerfile
# =========================================
# syntax=docker/dockerfile:1.6
FROM rust:1.75-bookworm as builder
WORKDIR /app
RUN apt-get update && apt-get install -y pkg-config libssl-dev && rm -rf /var/lib/apt/lists/*
COPY Cargo.toml Cargo.lock ./
COPY src ./src
COPY config ./config
COPY .sqlx ./\.sqlx
RUN --mount=type=cache,target=/usr/local/cargo/registry \
    --mount=type=cache,target=/app/target \
    cargo build --release

FROM debian:bookworm-slim
RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY --from=builder /app/target/release/ulink /usr/local/bin/ulink
COPY config ./config
EXPOSE 8080
ENV RUST_LOG=info
CMD ["/usr/local/bin/ulink"]

# =========================================
# File: docker-compose.yml
# =========================================
version: "3.9"
services:
  ulink:
    build: .
    image: ulink:latest
    ports:
      - "8080:8080"
    environment:
      - ULINK__DATABASE__URL=postgres://postgres:postgres@db:5432/ulink
      - ULINK__SERVER__PUBLIC_BASE=http://localhost:8080
      - ULINK__OBSERVABILITY__TRACING_LEVEL=info
      - ULINK__RATE_LIMIT__PER_MINUTE=200
      - ULINK__REDIS__ENABLED=false
    depends_on:
      - db
    ulimits:
      nofile: 65536
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: ulink
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
volumes:
  pgdata: {}

# =========================================
# File: k6/redirect.js
# =========================================
import http from 'k6/http';
import { check, sleep } from 'k6';
import { Counter, Trend } from 'k6/metrics';

export let options = {
  vus: 200,
  duration: '60s',
  thresholds: {
    http_req_failed: ['rate<0.01'],
    http_req_duration: ['p(99)<10'],
  },
};

const url = __ENV.SHORT_URL || 'http://localhost:8080/abcd123';

export default function () {
  let res = http.get(url, { redirects: 0 });
  check(res, {
    'status is 302': r => r.status === 302 || r.status === 301
  });
  sleep(0.001);
}

# =========================================
# File: k6/shorten.js
# =========================================
import http from 'k6/http';
import { sleep, check } from 'k6';

export let options = {
  vus: 50,
  duration: '30s',
  thresholds: {
    http_req_failed: ['rate<0.02'],
    http_req_duration: ['p(99)<60'],
  },
};

export default function () {
  const payload = JSON.stringify({ url: "https://example.com/path?q=1" });
  let res = http.post('http://localhost:8080/v1/shorten', payload, {
    headers: { 'Content-Type': 'application/json' }
  });
  check(res, { '200': r => r.status === 200 });
  sleep(0.05);
}

# =========================================
# File: README.md
# =========================================
# ulink - High-concurrency URL shortener in Rust

- REST:
  - POST /v1/shorten
  - GET /{code}
  - DELETE /v1/short/{code}
  - GET /v1/metrics

Run locally:
- cargo run --release
- curl -XPOST http://localhost:8080/v1/shorten -d '{"url":"https://example.com"}' -H 'content-type: application/json'

Production:
- docker compose up --build -d

# Bench (reference on 4c, cached redirects):
- GET p99: < 10ms
- POST cold p99: < 60ms

代码说明

• 功能描述：
  • 提供高并发URL短链服务，支持短码生成(ULID+Base62)、自定义别名唯一性校验与冲突重试、302跳转与匿名统计(UA/IP哈希/时间)、幂等删除、Prometheus指标导出(QPS、错误率、延迟直方图、缓存命中数、限流数、过期清理计数)。
  • 事务化写入：短链创建使用单条INSERT语句包裹在事务中，唯一约束冲突时重试（随机短码），自定义别名冲突返回409。
  • 缓存：热点缓存使用DashMap本地缓存，命中短路数据库查询；过期清理协程周期扫除已过期条目。
  • 速率限制：按IP按分钟阈值（InMemory实现，支持特性flag接入Redis版，失败默认放行），对shorten与resolve路径分别计数。
  • TTL过期：DB层保存expire_at，redirect时判定；后台任务按批次删除DB中过期数据，同时触发缓存逐出。
  • 安全校验：URL scheme仅允许http/https；可选域名白名单；自定义别名格式校验；无unsafe代码。
  • 观察性：tracing结构化日志、Prometheus指标、中间件记录请求计数/错误/延迟；/v1/metrics导出指标。
  • 容器化：多阶段构建，运行镜像基于debian-slim，体积<200MB；提供docker-compose整合Postgres/Redis。
  • 数据一致性：DB唯一主键code保障；冲突重试；命中异步统计与命中数累加（命中数更新为尽力而为，不阻塞跳转）。
• 参数说明：
  • 配置(config/default.yaml或环境变量ULINK__...):
    • server.bind：监听地址
    • server.public_base：短链域，用于返回short_url
    • database.url：PostgreSQL或SQLite连接串（支持sqlx Any）
    • cache.enabled/ttl_seconds：本地缓存开关与TTL
    • rate_limit.enabled/per_minute：每IP每分钟限额
    • security.allow_custom_alias/alias_regex/whitelist_domains：别名开关、格式与白名单
    • cleanup.interval_seconds/batch_limit：过期清理周期与批量上限
    • observability.log_json/metrics_path/tracing_level：日志与指标路径
    • redis.enabled/url：启用Redis限流时的连接信息（需启用redis特性）
• 返回值：
  • POST /v1/shorten：{code, short_url, expire_at}
  • GET /{code}：302跳转，无内容
  • DELETE /v1/short/{code}：{code, deleted}
  • GET /v1/metrics：Prometheus文本格式

使用示例

# 1) 启动服务（SQLite）
cargo run --release

# 2) 创建短链
curl -s -X POST http://localhost:8080/v1/shorten \
  -H 'content-type: application/json' \
  -d '{"url":"https://example.com/products/123?q=ok","ttl":86400}'

# 3) 使用自定义别名
curl -s -X POST http://localhost:8080/v1/shorten \
  -H 'content-type: application/json' \
  -d '{"url":"https://example.org","alias":"my-link_01"}'

# 4) 访问短链
curl -I http://localhost:8080/my-link_01

# 5) 删除短链（幂等）
curl -s -X DELETE http://localhost:8080/v1/short/my-link_01

# 6) 指标
curl -s http://localhost:8080/v1/metrics

# 7) Docker部署
docker compose up --build -d

# 8) K6重定向基准
SHORT_URL=http://localhost:8080/<your_code> k6 run k6/redirect.js

优化建议

• 性能考虑：
  • 将跳转路径尽量命中本地缓存：DashMap读为无锁读视图，常见热点可达>50k QPS，p99<10ms；可增加二级缓存(可选Redis)降低冷miss延迟。
  • 批量写入统计与命中计数：当前异步批处理1000条/200ms，可根据QPS动态调优缓冲大小和周期。
  • SQL优化：为code、expire_at建索引；Postgres生产环境建议使用UPSERT与CTE批量删除过期数据以减少往返。
  • 连接池：根据CPU与DB性能设置sqlx池大小（例如32）；将actix workers设置为与CPU核数一致。
  • 直方图桶：当前桶覆盖微秒到1秒范围；生产中可按SLA调整，减少指标开销，目标控制在<10%。
• 扩展性：
  • 多租户与鉴权：在short_urls增加tenant_id并对各API做鉴权与配额隔离。
  • 批量导入：新增批量shorten接口，事务化批量插入并返回映射列表；client侧可流式并发。
  • 分片与扩容：按code前缀或一致性哈希拆分到多个分片库/Redis分区；Nginx/Envoy做前端路由。
  • 布隆过滤器防碰撞：在生成短码前用布隆过滤器快速判断可能存在，降低DB冲突概率；冲突时重试。
  • 二级缓存：本地+Redis多级缓存，回源DB前先查Redis；失效采用异步通知或基于版本的淘汰。
  • 可观测性增强：接入OpenTelemetry/OTLP并上报到Prometheus/Grafana/Tempo；采样率/字段可配置。
• 注意事项：
  • 安全：若提供自定义别名，必须限制格式和长度，防止路径穿越与SEO欺骗；建议对public_base进行严格配置。
  • 速率限制：内存限流为单实例，若多副本部署需使用Redis限流或全局限流服务；fail-open策略可改为fail-close视业务需求。
  • 过期策略：DELETE LIMIT在Postgres不可直接用，已在仓储层做兼容选择；生产建议使用单独SQL脚本为不同DB优化。
  • 一致性：命中计数为尽力而为；如果需要强一致可改为消息队列/日志流(如Kafka)+定时汇总。
  • 容器化：生产建议使用distroless或alpine-musl静态链接以进一步缩小镜像；请确保CA证书存在以支持https回源。
  • 配置与密钥：如需对IP哈希加入私有盐，请通过环境变量注入并避免写入日志。
  • 测试：建议增加故障注入（DB/Redis超时、网络抖动）和端到端测试；使用k6与wrk对不同命中率进行压测，并校验p99与错误率阈值。