#!/usr/bin/env python
#
# Copyright 2016 Confluent Inc.
#
# Licensed 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.
#


#
# derived from https://github.com/verisign/python-confluent-schemaregistry.git
#

import sys
import json
import re

from threading import Thread, Event

from tests.avro.mock_schema_registry_client import MockSchemaRegistryClient
from confluent_kafka import avro
from confluent_kafka.avro.error import ClientError


if sys.version_info[0] < 3:
    import BaseHTTPServer as HTTPSERVER
else:
    import http.server as HTTPSERVER


class ReqHandler(HTTPSERVER.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.0"

    def do_GET(self):
        self.server._run_routes(self)

    def do_POST(self):
        self.server._run_routes(self)

    def log_message(self, format, *args):
        pass


class MockServer(HTTPSERVER.HTTPServer, object):
    def __init__(self, *args, **kwargs):
        super(MockServer, self).__init__(*args, **kwargs)
        self.counts = {}
        self.registry = MockSchemaRegistryClient()
        self.schema_cache = {}
        self.all_routes = {
            'GET': [
                (r"/schemas/ids/(\d+)", 'get_schema_by_id'),
                (r"/subjects/(\w+)/versions/(\d+)", 'get_by_version'),
                (r"/subjects/(\w+)/versions/latest", 'get_latest')
            ],
            'POST': [
                (r"/subjects/(\w+)/versions", 'register'),
                (r"/subjects/(\w+)", 'get_version')
            ]
        }

    def _send_response(self, resp, status, body):
        resp.send_response(status)
        resp.send_header("Content-Type", "application/json")
        resp.end_headers()
        resp.wfile.write(json.dumps(body).encode())
        resp.finish()

    def _create_error(self, msg, status=400, err_code=1):
        return (status, {
            "error_code": err_code,
            "message": msg
        })

    def _run_routes(self, req):
        self.add_count((req.command, req.path))
        routes = self.all_routes.get(req.command, [])
        for r in routes:
            m = re.match(r[0], req.path)
            if m:
                func = getattr(self, r[1])
                status, body = func(req, m.groups())
                return self._send_response(req, status, body)

        # here means we got a bad req
        status, body = self._create_error("bad path specified")
        self._send_response(req, status, body)

    def get_schema_by_id(self, req, groups):
        schema_id = int(groups[0])
        schema = self.registry.get_by_id(schema_id)
        if not schema:
            return self._create_error("schema not found", 404)
        result = {
            "schema": str(schema)
        }
        return (200, result)

    def _get_identity_schema(self, avro_schema):
        # normalized
        schema_str = str(avro_schema)
        if schema_str in self.schema_cache:
            return self.schema_cache[schema_str]
        self.schema_cache[schema_str] = avro_schema
        return avro_schema

    def _get_schema_from_body(self, req):
        length = int(req.headers['content-length'])
        data = req.rfile.read(length)
        data = json.loads(data.decode("utf-8"))
        schema = data.get("schema", None)
        if not schema:
            return None
        try:
            avro_schema = avro.loads(schema)
            return self._get_identity_schema(avro_schema)
        except ClientError:
            return None

    def register(self, req, groups):
        avro_schema = self._get_schema_from_body(req)
        if not avro_schema:
            return self._create_error("Invalid avro schema", 422, 42201)
        subject = groups[0]
        schema_id = self.registry.register(subject, avro_schema)
        return (200, {'id': schema_id})

    def check_registration(self, req, groups):
        avro_schema = self._get_schema_from_body(req)
        subject = groups[0]
        schema_id = self.registry.check_registration(subject, avro_schema)
        return (200, {'id': schema_id})

    def get_version(self, req, groups):
        avro_schema = self._get_schema_from_body(req)
        if not avro_schema:
            return self._create_error("Invalid avro schema", 422, 42201)
        subject = groups[0]
        version = self.registry.get_version(subject, avro_schema)
        if version == -1:
            return self._create_error("Not found", 404)
        schema_id = self.registry.get_id_for_schema(subject, avro_schema)

        result = {
            "schema": str(avro_schema),
            "subject": subject,
            "id": schema_id,
            "version": version
        }
        return (200, result)

    def get_latest(self, req, groups):
        subject = groups[0]
        schema_id, avro_schema, version = self.registry.get_latest_schema(subject)
        if schema_id is None:
            return self._create_error("Not found", 404)
        result = {
            "schema": str(avro_schema),
            "subject": subject,
            "id": schema_id,
            "version": version
        }
        return (200, result)

    def get_by_version(self, req, groups):
        subject = groups[0]
        version = int(groups[1])

        schema_id, avro_schema, version = self.registry.get_by_version(subject, version)
        if schema_id is None:
            return self._create_error("Not found", 404)
        result = {
            "schema": json.dumps(avro_schema.to_json()),
            "subject": subject,
            "id": schema_id,
            "version": version
        }
        return (200, result)

    def add_count(self, path):
        if path not in self.counts:
            self.counts[path] = 0
        self.counts[path] += 1


class ServerThread(Thread):
    def __init__(self, port):
        Thread.__init__(self)
        self.server = None
        self.port = port
        self.daemon = True
        self.started = Event()

    def run(self):
        self.server = MockServer(('127.0.0.1', self.port), ReqHandler)
        self.started.set()
        self.server.serve_forever()

    def start(self):
        """Start, and wait for server to be fully started, before returning."""
        super(ServerThread, self).start()
        self.started.wait()

    def shutdown(self):
        if self.server:
            self.server.shutdown()
            self.server.socket.close()


if __name__ == '__main__':
    s = ServerThread(0)
    s.start()