stack/apps/gateway/src/agent/connector-lease.service.ts

import { ForbiddenException, Inject, Injectable } from '@nestjs/common';
import { ConnectorLeaseCoordinator, normalizeConnectorLease } from '@mosaicstack/agent';
import {
  normalizeConnectorId,
  normalizeConnectorScopes,
  normalizeCorrelationId,
  normalizeLogicalAgentIdentity,
  normalizeLogicalBindingId,
  type AcquireConnectorLeaseInput,
  type ConnectorExecutionGrant,
  type ConnectorLease,
  type ConnectorLeaseAuditEvent,
  type FencedConnectorAdapter,
} from '@mosaicstack/types';
import type { ActorTenantScope } from '../auth/session-scope.js';
import { ConnectorLeaseRepository } from './connector-lease.repository.js';

export const CONNECTOR_LEASE_POLICY = Symbol('CONNECTOR_LEASE_POLICY');

export type ConnectorLeasePolicyAction =
  | 'lease.acquire'
  | 'lease.takeover'
  | 'lease.heartbeat'
  | 'lease.release'
  | 'lease.read'
  | 'grant.issue';

export interface ConnectorLeaseRequestContext {
  readonly actorScope: ActorTenantScope;
  readonly correlationId: string;
}

export interface GatewayConnectorLeaseRequest {
  readonly logicalAgentId: string;
  readonly bindingId: string;
  readonly connectorId: string;
  readonly scopes: readonly string[];
  readonly ttlMs: number;
}

export interface GatewayConnectorLeaseTakeoverRequest extends GatewayConnectorLeaseRequest {
  readonly expectedEpoch: string;
}

export interface GatewayConnectorGrantRequest {
  readonly lease: ConnectorLease;
  readonly scopes: readonly string[];
  readonly ttlMs: number;
}

export interface ConnectorLeasePolicySubject {
  readonly action: ConnectorLeasePolicyAction;
  readonly actorId: string;
  readonly tenantId: string;
  readonly logicalAgentId: string;
  readonly bindingId: string;
  readonly connectorId: string;
  readonly requestedScopes: readonly string[];
  readonly requestedTtlMs: number | null;
}

export interface ConnectorLeasePolicy {
  authorize(subject: ConnectorLeasePolicySubject): Promise<boolean>;
}

/** M1 has no concrete cutover policy: unconfigured production use fails closed. */
@Injectable()
export class DenyConnectorLeasePolicy implements ConnectorLeasePolicy {
  async authorize(_subject: ConnectorLeasePolicySubject): Promise<boolean> {
    return false;
  }
}

/** Gateway-owned policy surface for durable connector authority and fenced effects. */
@Injectable()
export class ConnectorLeaseService {
  private readonly coordinator: ConnectorLeaseCoordinator;

  constructor(
    @Inject(ConnectorLeaseRepository) private readonly repository: ConnectorLeaseRepository,
    @Inject(CONNECTOR_LEASE_POLICY) private readonly policy: ConnectorLeasePolicy,
  ) {
    this.coordinator = new ConnectorLeaseCoordinator(repository);
  }

  async acquire(
    request: GatewayConnectorLeaseRequest,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorLease> {
    const command = this.command(request, context);
    await this.assertPolicy('lease.acquire', command, context, command.scopes, command.ttlMs);
    return this.coordinator.acquire({ ...command, correlationId: this.correlation(context) });
  }

  async takeover(
    request: GatewayConnectorLeaseTakeoverRequest,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorLease> {
    const command = this.command(request, context);
    await this.assertPolicy('lease.takeover', command, context, command.scopes, command.ttlMs);
    return this.coordinator.takeover({
      ...command,
      expectedEpoch: request.expectedEpoch,
      correlationId: this.correlation(context),
    });
  }

  async heartbeat(
    lease: ConnectorLease,
    ttlMs: number,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorLease> {
    const normalizedLease = normalizeConnectorLease(lease);
    const durableLease = await this.durableLifecycleLease(normalizedLease, context);
    await this.assertPolicy('lease.heartbeat', durableLease, context, durableLease.scopes, ttlMs);
    return this.coordinator.heartbeat({
      lease: durableLease,
      ttlMs,
      correlationId: this.correlation(context),
    });
  }

  async release(lease: ConnectorLease, context: ConnectorLeaseRequestContext): Promise<void> {
    const normalizedLease = normalizeConnectorLease(lease);
    const durableLease = await this.durableLifecycleLease(normalizedLease, context);
    await this.assertPolicy('lease.release', durableLease, context, durableLease.scopes, null);
    await this.coordinator.release({
      lease: durableLease,
      correlationId: this.correlation(context),
    });
  }

  async current(
    logicalAgentId: string,
    bindingId: string,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorLease | null> {
    const binding = {
      identity: normalizeLogicalAgentIdentity({
        tenantId: context.actorScope.tenantId,
        logicalAgentId,
      }),
      bindingId: normalizeLogicalBindingId(bindingId),
      connectorId: 'gateway',
    };
    await this.assertPolicy('lease.read', binding, context, [], null);
    return this.coordinator.current(binding);
  }

  async issueGrant(
    request: GatewayConnectorGrantRequest,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorExecutionGrant> {
    const lease = normalizeConnectorLease(request.lease);
    await this.assertTenant(lease, context);
    const scopes = normalizeConnectorScopes(request.scopes);
    await this.assertPolicy('grant.issue', lease, context, scopes, request.ttlMs);
    return this.coordinator.issueGrant({
      lease,
      scopes,
      ttlMs: request.ttlMs,
      correlationId: this.correlation(context),
    });
  }

  async executeGrant<TInput, TOutput>(
    grant: ConnectorExecutionGrant,
    requiredScope: string,
    input: TInput,
    adapter: FencedConnectorAdapter<TInput, TOutput>,
  ): Promise<TOutput> {
    return this.coordinator.executeGrant(grant, requiredScope, input, adapter);
  }

  private command(
    request: GatewayConnectorLeaseRequest,
    context: ConnectorLeaseRequestContext,
  ): Omit<AcquireConnectorLeaseInput, 'correlationId'> {
    return {
      identity: normalizeLogicalAgentIdentity({
        tenantId: context.actorScope.tenantId,
        logicalAgentId: request.logicalAgentId,
      }),
      bindingId: normalizeLogicalBindingId(request.bindingId),
      connectorId: normalizeConnectorId(request.connectorId),
      scopes: normalizeConnectorScopes(request.scopes),
      ttlMs: request.ttlMs,
    };
  }

  private async assertTenant(
    lease: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
    context: ConnectorLeaseRequestContext,
  ): Promise<void> {
    if (lease.identity.tenantId !== context.actorScope.tenantId) {
      await this.recordPolicyDenial(
        {
          identity: {
            tenantId: context.actorScope.tenantId,
            logicalAgentId: 'untrusted',
          },
          bindingId: 'untrusted',
          connectorId: 'untrusted',
        },
        context,
      );
      throw new ForbiddenException('Connector authority tenant scope denied');
    }
  }

  private async durableLifecycleLease(
    submittedLease: ConnectorLease,
    context: ConnectorLeaseRequestContext,
  ): Promise<ConnectorLease> {
    await this.assertTenant(submittedLease, context);
    const durableLease = await this.coordinator.current(submittedLease);
    if (!durableLease || !hasSameLifecycleAuthority(submittedLease, durableLease)) {
      await this.recordPolicyDenial(durableLease ?? submittedLease, context);
      throw new ForbiddenException('Connector authority policy denied');
    }
    return durableLease;
  }

  private async assertPolicy(
    action: ConnectorLeasePolicyAction,
    subject: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
    context: ConnectorLeaseRequestContext,
    requestedScopes: readonly string[],
    requestedTtlMs: number | null,
  ): Promise<void> {
    const allowed = await this.policy.authorize({
      action,
      actorId: context.actorScope.userId,
      tenantId: subject.identity.tenantId,
      logicalAgentId: subject.identity.logicalAgentId,
      bindingId: subject.bindingId,
      connectorId: subject.connectorId,
      requestedScopes: Object.freeze([...requestedScopes]),
      requestedTtlMs,
    });
    if (!allowed) {
      await this.recordPolicyDenial(subject, context);
      throw new ForbiddenException('Connector authority policy denied');
    }
  }

  private async recordPolicyDenial(
    subject: Pick<ConnectorLease, 'identity' | 'bindingId' | 'connectorId'>,
    context: ConnectorLeaseRequestContext,
  ): Promise<void> {
    const event: ConnectorLeaseAuditEvent = {
      identity: subject.identity,
      bindingId: subject.bindingId,
      connectorId: subject.connectorId,
      event: 'reject',
      outcome: 'denied',
      reason: 'policy_denied',
      correlationId: this.correlation(context),
      occurredAt: new Date().toISOString(),
    };
    await this.repository.recordAudit(event);
  }

  private correlation(context: ConnectorLeaseRequestContext): string {
    return normalizeCorrelationId(context.correlationId);
  }
}

function hasSameLifecycleAuthority(
  submittedLease: ConnectorLease,
  durableLease: ConnectorLease,
): boolean {
  return (
    submittedLease.identity.tenantId === durableLease.identity.tenantId &&
    submittedLease.identity.logicalAgentId === durableLease.identity.logicalAgentId &&
    submittedLease.bindingId === durableLease.bindingId &&
    submittedLease.leaseId === durableLease.leaseId &&
    submittedLease.connectorId === durableLease.connectorId &&
    submittedLease.leaseEpoch === durableLease.leaseEpoch &&
    submittedLease.scopes.length === durableLease.scopes.length &&
    submittedLease.scopes.every((scope: string, index: number): boolean => {
      return scope === durableLease.scopes[index];
    })
  );
}