import { getEnv } from '../../../env/env'
import * as pubsub from '@google-cloud/pubsub'
import { v4 } from 'uuid'
import { defer } from 'bluebird'

jest.setTimeout(1000000)

test('errors if messages are inflight while closing, then deleting subscription causing it to fail', async () => {
  const client = createPubSubSdk(getEnv())
  // with enough iterations we should be able to trigger it.
  // NOTE: when the fix is applied, this succeeds.
  for (let i = 0; i < 100; i++) {
    const topicName = `bug-topic-${v4()}`
    const subName = `bug-sub-${v4()}`
    const [topic] = await client.createTopic(topicName)
    const [sub] = await topic.createSubscription(subName, {
      expirationPolicy: { ttl: { seconds: 60 } },
    })

    await topic.publish(Buffer.from(JSON.stringify({ hello: 'world' })))

    const onHandlerEnter = defer()
    const onError = defer()
    sub.setOptions({ batching: { maxMilliseconds: 100 } })
    sub.addListener('message', (message: pubsub.Message) => {
      message.ack()
      onHandlerEnter.resolve()
    })
    sub.addListener('error', (err) => {
      console.log(err)
      onError.reject(err)
    })
    sub.open()

    await onHandlerEnter.promise

    // Kicks off the flush
    await new Promise((resolve) => setTimeout(resolve, 100))

    // Close while requests are pending. Wait for it.
    await sub.close()
    // Delete the subscription. This will trigger an error in ack.
    await sub.delete()
    await topic.delete()
    await Promise.race([
      new Promise((resolve) => setTimeout(resolve, 1000)),
      onError.promise,
    ])
  }
})

function createPubSubSdk(env: any) {
  return new pubsub.PubSub({
    projectId: env.GCLOUD_PROJECT,
    credentials: JSON.parse(env.GCLOUD_CREDENTIALS),
  })
}