/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 */

/*
 * GraphBinaryReader v4 response format tests.
 * Tests the reader's ability to parse v4 response format:
 * {version:0x84}{bulked:Byte}{result_data stream}{marker:0xFD 0x00 0x00}{status_code:Int bare}{status_message:nullable}{exception:nullable}
 */

import { assert } from 'chai';
import { Buffer } from 'buffer';
import GraphBinaryReader from '../../../lib/structure/io/binary/internals/GraphBinaryReader.js';
import StreamReader from '../../../lib/structure/io/binary/internals/StreamReader.js';
import ioc from '../../../lib/structure/io/binary/GraphBinary.js';
import { Traverser } from '../../../lib/process/traversal.js';
import ResponseError from '../../../lib/driver/response-error.js';

describe('GraphBinaryReader', () => {
  const reader = new GraphBinaryReader(ioc);

  describe('input validation', () => {
    it('undefined buffer throws error', async () => {
      try {
        await reader.readResponse(undefined);
        assert.fail('should have thrown');
      } catch (e) {
        assert.match(e.message, /Buffer is missing/);
      }
    });

    it('null buffer throws error', async () => {
      try {
        await reader.readResponse(null);
        assert.fail('should have thrown');
      } catch (e) {
        assert.match(e.message, /Buffer is missing/);
      }
    });

    it('empty buffer throws error', async () => {
      try {
        await reader.readResponse(Buffer.alloc(0));
        assert.fail('should have thrown');
      } catch (e) {
        assert.match(e.message, /Buffer is empty/);
      }
    });
  });

  describe('version validation', () => {
    it('rejects version 0x00', async () => {
      try {
        await reader.readResponse(Buffer.from([0x00, 0x00]));
        assert.fail('should have thrown');
      } catch (e) {
        assert.match(e.message, /Unsupported version/);
      }
    });

    it('rejects version 0x81', async () => {
      try {
        await reader.readResponse(Buffer.from([0x81, 0x00]));
        assert.fail('should have thrown');
      } catch (e) {
        assert.match(e.message, /Unsupported version/);
      }
    });
  });

  describe('non-bulked responses', () => {
    it('single value', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int: type_code=0x01, value_flag=0x00, value=67
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null flag
        0x01  // exception null flag
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 200, message: null, exception: null },
        result: { data: [67], bulked: false },
      });
    });

    it('multiple values', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int: 67
        0x03, 0x00, 0x00, 0x00, 0x00, 0x05, 0x68, 0x65, 0x6C, 0x6C, 0x6F, // fq String: "hello"
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 200, message: null, exception: null },
        result: { data: [67, 'hello'], bulked: false }
      });
    });

    it('empty result', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0xFD, 0x00, 0x00, // marker (no data)
        0x00, 0x00, 0x00, 0xCC, // status_code=204
        0x01, // status_message null
        0x01  // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 204, message: null, exception: null },
        result: { data: [], bulked: false }
      });
    });
  });

  describe('bulked responses', () => {
    it('single item with bulk count', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x01, // bulked=true
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int: 67
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, // fq Long bulk=3
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 200, message: null, exception: null },
        result: { data: [{ v: 67, bulk: 3 }], bulked: true }
      });
    });

    it('multiple items with bulk counts', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x01, // bulked=true
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int: 67
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, // fq Long bulk=2
        0x03, 0x00, 0x00, 0x00, 0x00, 0x05, 0x68, 0x65, 0x6C, 0x6C, 0x6F, // fq String: "hello"
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, // fq Long bulk=1
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 200, message: null, exception: null },
        result: { data: [{ v: 67, bulk: 2 }, { v: 'hello', bulk: 1 }], bulked: true }
      });
    });
  });

  describe('status codes', () => {
    it('status 403', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x01, 0x93, // status_code=403
        0x01, 0x01 // null message, null exception
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.code, 403);
    });

    it('status 500', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x01, 0xF4, // status_code=500
        0x01, 0x01 // null message, null exception
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.code, 500);
    });
  });

  describe('nullable status_message', () => {
    it('present message', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x00, // message present flag
        0x00, 0x00, 0x00, 0x07, 0x53, 0x75, 0x63, 0x63, 0x65, 0x73, 0x73, // bare String: "Success"
        0x01 // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.message, 'Success');
    });

    it('null message', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // message null flag
        0x01 // exception null
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.message, null);
    });
  });

  describe('nullable exception', () => {
    it('present exception', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x01, 0xF4, // status_code=500
        0x01, // message null
        0x00, // exception present flag
        0x00, 0x00, 0x00, 0x05, 0x45, 0x72, 0x72, 0x6F, 0x72 // bare String: "Error"
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.exception, 'Error');
    });

    it('null exception', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // message null
        0x01 // exception null flag
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.status.exception, null);
    });
  });

  describe('error response', () => {
    it('no result data with error status', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0xFD, 0x00, 0x00, // marker (no data)
        0x00, 0x00, 0x01, 0xF4, // status_code=500
        0x00, // message present
        0x00, 0x00, 0x00, 0x0E, 0x49, 0x6E, 0x74, 0x65, 0x72, 0x6E, 0x61, 0x6C, 0x20, 0x65, 0x72, 0x72, 0x6F, 0x72, // "Internal error" (14 chars)
        0x00, // exception present
        0x00, 0x00, 0x00, 0x09, 0x45, 0x78, 0x63, 0x65, 0x70, 0x74, 0x69, 0x6F, 0x6E // "Exception"
      ]);
      const result = await reader.readResponse(buffer);
      assert.deepEqual(result, {
        status: { code: 500, message: 'Internal error', exception: 'Exception' },
        result: { data: [], bulked: false }
      });
    });
  });

  describe('complex result values', () => {
    it('vertex in result data', async () => {
      const buffer = Buffer.from([
        0x84, 0x00, // version, bulked=false
        0x11, 0x00, // fq Vertex: type_code=0x11, value_flag=0x00
        0x01, 0x00, 0x00, 0x00, 0x00, 0x01, // id: fq Int=1
        0x00, 0x00, 0x00, 0x01, // label: bare List length=1
        0x03, 0x00, 0x00, 0x00, 0x00, 0x06, 0x70, 0x65, 0x72, 0x73, 0x6F, 0x6E, // fq String="person"
        0x0B, 0x00, 0x00, 0x00, 0x00, 0x00, // properties: fq List=empty
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, 0x01 // null message, null exception
      ]);
      const result = await reader.readResponse(buffer);
      assert.equal(result.result.data.length, 1);
      assert.equal(result.result.data[0].id, 1);
      assert.equal(result.result.data[0].label, 'person');
    });
  });

  describe('readResponseStream', () => {
    it('non-bulked stream yields raw values', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const results = [];
      for await (const item of reader.readResponseStream(streamReader)) {
        results.push(item);
      }
      assert.deepEqual(results, [67]);
    });

    it('bulked stream yields Traverser objects', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x01, // bulked=true
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, // fq Long=3
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const results = [];
      for await (const item of reader.readResponseStream(streamReader)) {
        results.push(item);
      }
      assert.equal(results.length, 1);
      assert.instanceOf(results[0], Traverser);
      assert.equal(results[0].object, 67);
      assert.equal(results[0].bulk, 3);
    });

    it('bulked stream Traversers expand correctly via Traversal', async () => {
      const { Traversal } = await import('../../../lib/process/traversal.js');
      const buffer = Buffer.from([
        0x84, // version
        0x01, // bulked=true
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03, // fq Long=3
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, 0x01
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const gen = reader.readResponseStream(streamReader);
      const traversal = new Traversal(null, null);
      traversal._resultsStream = gen;
      const list = await traversal.toList();
      // bulk=3 should expand to 3 copies of 67
      assert.deepEqual(list, [67, 67, 67]);
    });

    it('trailing status is accessible via Traversal.getStatus()', async () => {
      const { Traversal } = await import('../../../lib/process/traversal.js');
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x00, // message present
        0x00, 0x00, 0x00, 0x02, 0x4F, 0x4B, // "OK"
        0x01  // exception null
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const gen = reader.readResponseStream(streamReader);
      const traversal = new Traversal(null, null);
      traversal._resultsStream = gen;

      assert.isNull(traversal.getStatus());
      await traversal.toList();
      const status = traversal.getStatus();
      assert.isNotNull(status);
      assert.equal(status.code, 200);
      assert.equal(status.message, 'OK');
      assert.isNull(status.exception);
    });

    it('throws ResponseError on server error status after yielding values', async () => {
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x01, 0xF4, // status_code=500
        0x00, // message present
        0x00, 0x00, 0x00, 0x0E, 0x49, 0x6E, 0x74, 0x65, 0x72, 0x6E, 0x61, 0x6C, 0x20, 0x65, 0x72, 0x72, 0x6F, 0x72, // "Internal error"
        0x00, // exception present
        0x00, 0x00, 0x00, 0x09, 0x45, 0x78, 0x63, 0x65, 0x70, 0x74, 0x69, 0x6F, 0x6E // "Exception"
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const results = [];
      try {
        for await (const item of reader.readResponseStream(streamReader)) {
          results.push(item);
        }
        assert.fail('should have thrown');
      } catch (e) {
        assert.instanceOf(e, ResponseError);
        assert.equal(e.statusCode, 500);
        assert.match(e.message, /Internal error/);
      }
      // Verify partial results were yielded before the error
      assert.deepEqual(results, [67]);
    });

    it('hasNext() peeks without consuming from streaming source', async () => {
      const { Traversal } = await import('../../../lib/process/traversal.js');
      const buffer = Buffer.from([
        0x84, 0x00,
        0x01, 0x00, 0x00, 0x00, 0x00, 0x01, // fq Int=1
        0x01, 0x00, 0x00, 0x00, 0x00, 0x02, // fq Int=2
        0xFD, 0x00, 0x00,
        0x00, 0x00, 0x00, 0xC8,
        0x01, 0x01
      ]);
      const streamReader = StreamReader.fromBuffer(buffer);
      const gen = reader.readResponseStream(streamReader);
      const traversal = new Traversal(null, null);
      traversal._resultsStream = gen;

      // hasNext should return true without consuming
      assert.isTrue(await traversal.hasNext());
      assert.isTrue(await traversal.hasNext()); // calling again should still be true

      // next() should return the peeked value
      const first = await traversal.next();
      assert.equal(first.value, 1);

      assert.isTrue(await traversal.hasNext());
      const second = await traversal.next();
      assert.equal(second.value, 2);

      assert.isFalse(await traversal.hasNext());
    });

    it('deserializes correctly when response is split into 1-byte chunks via ReadableStream', async () => {
      // Build a complete v4 response: version + bulked=false + Int(67) + String("hi") + marker + status 200
      const buffer = Buffer.from([
        0x84, // version
        0x00, // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x43, // fq Int=67
        0x03, 0x00, 0x00, 0x00, 0x00, 0x02, 0x68, 0x69, // fq String="hi"
        0xFD, 0x00, 0x00, // marker
        0x00, 0x00, 0x00, 0xC8, // status_code=200
        0x01, // status_message null
        0x01  // exception null
      ]);
      // Split into 1-byte chunks
      const chunks = [];
      for (let i = 0; i < buffer.length; i++) {
        chunks.push(buffer.subarray(i, i + 1));
      }
      const stream = new ReadableStream({
        pull(controller) {
          if (chunks.length > 0) {
            controller.enqueue(new Uint8Array(chunks.shift()));
          } else {
            controller.close();
          }
        },
      });
      const streamReader = StreamReader.fromReadableStream(stream);
      const results = [];
      for await (const item of reader.readResponseStream(streamReader)) {
        results.push(item);
      }
      assert.deepEqual(results, [67, 'hi']);
    });

    it('yields first result before remaining data is available', async () => {
      // v4 response: version + bulked=false + Int(42) + Int(99) + marker + status 200
      const response = Buffer.from([
        0x84,                                 // version
        0x00,                                 // bulked=false
        0x01, 0x00, 0x00, 0x00, 0x00, 0x2A,  // fq Int=42
        0x01, 0x00, 0x00, 0x00, 0x00, 0x63,  // fq Int=99
        0xFD, 0x00, 0x00,                     // marker
        0x00, 0x00, 0x00, 0xC8,               // status_code=200
        0x01,                                 // status_message null
        0x01                                  // exception null
      ]);

      // Split: first chunk has header (2 bytes) + first value (6 bytes) = 8 bytes.
      // The rest (second value + marker + status) is withheld.
      const firstChunk = response.subarray(0, 8);
      const remaining = response.subarray(8);

      let releaseRemaining;
      const remainingReady = new Promise((resolve) => { releaseRemaining = resolve; });
      let closed = false;

      const stream = new ReadableStream({
        async pull(controller) {
          if (!closed) {
            // First pull: deliver just enough for the header + first value
            controller.enqueue(new Uint8Array(firstChunk));
            closed = true;
            // Block until test signals to release the rest
            await remainingReady;
            controller.enqueue(new Uint8Array(remaining));
            controller.close();
          }
        },
      });

      const streamReader = StreamReader.fromReadableStream(stream);
      const gen = reader.readResponseStream(streamReader);

      // First next() should resolve with 42 — only the first chunk was delivered
      const first = await gen.next();
      assert.isFalse(first.done);
      assert.equal(first.value, 42);

      // Release the remaining data
      releaseRemaining();

      // Second value should now be available
      const second = await gen.next();
      assert.isFalse(second.done);
      assert.equal(second.value, 99);

      // Generator should complete with status as return value
      const end = await gen.next();
      assert.isTrue(end.done);
      assert.equal(end.value.code, 200);
    });
  });
});