import { filter, pipe } from 'graphql-yoga'
import { Arg, Field, Int, ObjectType, Resolver, Root, Subscription } from 'type-graphql'

import { assertDomainSubscribeAllowed, pubsub, ScalarObject } from '@things-factory/shell'

import { TwinEngine } from '../../engine/index.js'
import { TWIN_STATE_TOPIC, twinStateFilter, twinStatePayload, type TwinStatePayload } from '../../engine/twin-state-channel.js'

/* State 채널 메시지 — snapshot | delta | clock 멀티플렉싱(커널 SubscriptionMessage 대응). */
@ObjectType({ description: 'Twin State channel message multiplexing snapshot | delta | clock.' })
export class TwinStateMessage {
  @Field({ description: 'Twin runtime instance id this message belongs to.' })
  instanceId: string

  @Field({ description: 'Message kind: snapshot | delta | clock.' })
  kind: string

  @Field(type => Int, { nullable: true, description: 'Monotonic state revision.' })
  revision?: number

  @Field(type => ScalarObject, { nullable: true, description: 'Payload — snapshot: StateSnapshot / delta: CanonicalEnvelope / clock: { simClockMs }.' })
  payload?: any
}

/*
 * 트윈 State 구독 — state-register/data-resolver 패턴 미러.
 * 구독 즉시 현재 snapshot 발행(process.nextTick) 후 delta/clock 스트림.
 *
 * The tenant is asked while streaming, not only at subscribe time (2026-09-24). Ownership used to be
 * checked once and the stream filtered on instanceId alone; the same id in two tenants is the normal
 * case (`runtime-key.ts`), so other tenants' deltas came through, and the snapshot sent to a new
 * subscriber reached every tenant subscribed to that id. Messages now carry their domain and the filter
 * compares it (`engine/twin-state-channel.ts`). A twin the caller does not own is refused rather than
 * given an empty stream — an empty stream cannot tell "nothing changed" from "you may not see this"
 * (shell's `data` throws too).
 */
@Resolver()
export class TwinStateSubscription {
  @Subscription(returns => TwinStateMessage, {
    subscribe: async ({ args, context }) => {
      const { instanceId } = args
      const { domain, user } = (context as any)?.state ?? {}
      await assertDomainSubscribeAllowed(domain, user, (d, u) => process.superUserGranted(d, u))

      /* One refusal for both cases, so the answer does not reveal whether another tenant has such a twin. */
      if (!TwinEngine.owns(domain.id, instanceId)) throw new Error(`twin instance not running in this tenant: ${instanceId}`)

      /* Current state first (a late subscriber still sees snapshot → delta). It reaches this tenant's subscribers only. */
      process.nextTick(() => {
        const snap = TwinEngine.snapshot(domain.id, instanceId)
        if (snap) {
          pubsub.publish(TWIN_STATE_TOPIC, twinStatePayload(domain.id, { instanceId, kind: 'snapshot', revision: snap.revision, payload: snap }))
        }
      })

      return pipe(pubsub.subscribe(TWIN_STATE_TOPIC), filter(twinStateFilter(domain.id, instanceId)))
    }
  })
  twinState(@Root() payload: TwinStatePayload, @Arg('instanceId') instanceId: string): TwinStateMessage {
    return payload.twinState as TwinStateMessage
  }
}