WIP server

5 files changed, 334 insertions, 23 deletions

diff --git a/src_py/hatter/backend.py b/src_py/hatter/backend.py
index 059c636..e89a867 100644
--- a/src_py/hatter/backend.py
+++ b/src_py/hatter/backend.py
@@ -1,12 +1,20 @@
 from pathlib import Path
+import typing
 
 from hat import aio
 
+from hatter import common
+
 
 async def create(db_path: Path
                  ) -> 'Backend':
     backend = Backend()
     backend._async_group = aio.Group()
+    backend._executor = aio.create_executor(1)
+
+    backend._db = await backend._executor(_ext_create, db_path)
+    backend.async_group.spawn(aio.call_on_cancel, backend._executor,
+                              _ext_close, backend._db)
 
     return backend
 
@@ -16,3 +24,50 @@ class Backend(aio.Resource):
     @property
     def async_group(self):
         return self._async_group
+
+    async def get_commits(self,
+                          repo: typing.Optional[str],
+                          statuses: typing.Optional[typing.Set[common.Status]],
+                          order: common.Order
+                          ) -> typing.List[common.Commit]:
+        return await self.async_group.spawn(
+            self._executor, _ext_get_commits, self._db, repo, statuses, order)
+
+    async def get_commit(self,
+                         repo: str,
+                         commit_hash: str
+                         ) -> typing.Optional[common.Commit]:
+        return await self.async_group.spawn(
+            self._executor, _ext_get_commit, self._db, repo, commit_hash)
+
+    async def update_commit(self, commit: common.Commit):
+        return await self._async_group.spawn(
+            self._executor, _ext_update_commit, self._db, commit)
+
+    async def remove_commit(self, commit: common.Commit):
+        return await self.async_group.spawn(
+            self._executor, _ext_remove_commit, self._db, commit)
+
+
+def _ext_create(db_path):
+    pass
+
+
+def _ext_close(db):
+    pass
+
+
+def _ext_get_commits(db, repo, statuses, order):
+    return []
+
+
+def _ext_get_commit(db, repo, commit_hash):
+    pass
+
+
+def _ext_update_commit(db, commit):
+    pass
+
+
+def _ext_remove_commit(db, commit):
+    pass
diff --git a/src_py/hatter/common.py b/src_py/hatter/common.py
index cf99bb9..1b1de47 100644
--- a/src_py/hatter/common.py
+++ b/src_py/hatter/common.py
@@ -1,4 +1,6 @@
 from pathlib import Path
+import enum
+import typing
 
 from hat import json
 
@@ -8,3 +10,23 @@ package_path: Path = Path(__file__).parent
 json_schema_repo: json.SchemaRepository = json.SchemaRepository(
     json.json_schema_repo,
     json.SchemaRepository.from_json(package_path / 'json_schema_repo.json'))
+
+
+class Order(enum.Enum):
+    ASC = 'ASC'
+    DESC = 'DESC'
+
+
+class Status(enum.Enum):
+    PENDING = 0
+    RUNNING = 1
+    SUCCESS = 2
+    FAILURE = 3
+
+
+class Commit(typing.NamedTuple):
+    repo: str
+    hash: str
+    change: float
+    status: Status
+    output: str
diff --git a/src_py/hatter/main.py b/src_py/hatter/main.py
index 1d094b5..53b2032 100644
--- a/src_py/hatter/main.py
+++ b/src_py/hatter/main.py
@@ -58,10 +58,10 @@ def main(log_level: str,
 
 @main.command()
 @click.argument('url', required=True)
-@click.argument('commit', required=False, default='master')
+@click.argument('ref', required=False, default='HEAD')
 @click.argument('action', required=False, default='.hatter.yaml')
 def execute(url: str,
-            commit: str,
+            ref: str,
             action: str):
     with contextlib.suppress(Exception):
         path = Path(url)
@@ -79,11 +79,11 @@ def execute(url: str,
                        cwd=str(repo_dir),
                        check=True)
 
-        subprocess.run(['git', 'fetch', '-q', '--depth=1', 'origin', commit],
+        subprocess.run(['git', 'fetch', '-q', '--depth=1', 'origin', ref],
                        cwd=str(repo_dir),
                        check=True)
 
-        subprocess.run(['git', 'checkout', '-q', commit],
+        subprocess.run(['git', 'checkout', '-q', 'FETCH_HEAD'],
                        cwd=str(repo_dir),
                        check=True)
 
@@ -135,6 +135,9 @@ async def async_server(host: str,
         server = await hatter.server.create(conf, backend)
         _bind_resource(async_group, server)
 
+        for repo in server.get_repos():
+            server.sync_repo(repo)
+
         ui = await hatter.ui.create(host, port, server)
         _bind_resource(async_group, ui)
 
diff --git a/src_py/hatter/server.py b/src_py/hatter/server.py
index f9ee29f..ccf100f 100644
--- a/src_py/hatter/server.py
+++ b/src_py/hatter/server.py
@@ -1,6 +1,12 @@
+import asyncio
+import multiprocessing
+import time
+import typing
+
 from hat import aio
 from hat import json
 
+from hatter import common
 import hatter.backend
 
 
@@ -8,8 +14,37 @@ async def create(conf: json.Data,
                  backend: hatter.backend.Backend
                  ) -> 'Server':
     server = Server()
+    server._conf = conf
     server._backend = backend
     server._async_group = aio.Group()
+    server._lock = asyncio.Lock()
+    server._run_queue = aio.Queue()
+    server._sync_events = {}
+
+    for repo, repo_conf in conf['repos'].items():
+        sync_event = asyncio.Event()
+        server._sync_events[repo] = sync_event
+        server.async_group.spawn(server._sync_loop, repo_conf, sync_event)
+
+    for _ in range(multiprocessing.cpu_count()):
+        server.async_group.spawn(server._run_loop)
+
+    try:
+        commits = await backend.get_commits(repo=None,
+                                            statuses={common.Status.PENDING,
+                                                      common.Status.RUNNING},
+                                            order=common.Order.ASC)
+
+        for commit in commits:
+            commit = commit._replace(change=time.time(),
+                                     status=common.Status.PENDING,
+                                     output='')
+            await backend.update_commit(commit)
+            server._run_queue.put_nowait(commit)
+
+    except BaseException:
+        await aio.uncancellable(server.async_close())
+        raise
 
     return server
 
@@ -19,3 +54,78 @@ class Server(aio.Resource):
     @property
     def async_group(self):
         return self._async_group
+
+    def get_repos(self) -> typing.Iterable[str]:
+        return self._conf['repos'].keys()
+
+    async def get_commits(self,
+                          repo: typing.Optional[str],
+                          ) -> typing.Iterable[common.Commit]:
+        if repo and repo not in self._conf['repos']:
+            raise ValueError(f'invalid repo {repo}')
+
+        commits = await self._backend.get_commits(repo=repo,
+                                                  statuses=None,
+                                                  order=common.Order.DESC)
+
+        return commits
+
+    async def get_commit(self,
+                         repo: str,
+                         commit_hash: str
+                         ) -> common.Commit:
+        if repo not in self._conf['repos']:
+            raise ValueError(f'invalid repo {repo}')
+
+        async with self._lock:
+            commit = await self._backend.get_commit(repo, commit_hash)
+
+            if not commit:
+                commit = common.Commit(repo=repo,
+                                       hash=commit_hash,
+                                       change=time.time(),
+                                       status=common.Status.PENDING,
+                                       output='')
+                await self._backend.update_commit(commit)
+                self._run_queue.put_nowait(commit)
+
+        return commit
+
+    def sync_repo(self, repo: str):
+        self._sync_events[repo].set()
+
+    async def rerun_commit(self,
+                           repo: str,
+                           commit_hash: str):
+        if repo not in self._conf['repos']:
+            raise ValueError(f'invalid repo {repo}')
+
+        async with self._lock:
+            commit = await self._backend.get_commit(repo, commit_hash)
+            if not commit:
+                raise ValueError(f'invalid commit {commit_hash}')
+
+            commit = commit._replace(change=time.time(),
+                                     status=common.Status.PENDING,
+                                     output='')
+            await self._backend.update_commit(commit)
+            self._run_queue.put_nowait(commit)
+
+    async def remove_commit(self,
+                            repo: str,
+                            commit_hash: str):
+        if repo not in self._conf['repos']:
+            raise ValueError(f'invalid repo {repo}')
+
+        async with self._lock:
+            commit = await self._backend.get_commit(repo, commit_hash)
+            if not commit:
+                raise ValueError(f'invalid commit {commit_hash}')
+
+            await self._backend.remove_commit(commit)
+
+    async def _sync_loop(self, repo_conf, sync_event):
+        pass
+
+    async def _run_loop(self):
+        pass
diff --git a/src_py/hatter/ui.py b/src_py/hatter/ui.py
index 525df6b..9368d63 100644
--- a/src_py/hatter/ui.py
+++ b/src_py/hatter/ui.py
@@ -1,4 +1,5 @@
 from pathlib import Path
+import datetime
 
 from hat import aio
 import aiohttp.web
@@ -19,18 +20,18 @@ async def create(host: str,
     ui._async_group = aio.Group()
 
     app = aiohttp.web.Application()
-    app.add_routes([
-        aiohttp.web.get('/', ui._get_root_handler),
-        aiohttp.web.get('/repo/{repo}', server._get_repo_handler),
-        aiohttp.web.get('/repo/{repo}/commit/{commit}',
-                        server._get_commit_handler),
-        aiohttp.web.post('/repo/{repo}/webhook',
-                         server._post_webhook_handler),
-        aiohttp.web.post('/repo/{repo}/commit/{commit}/run',
-                         server._post_run_handler),
-        aiohttp.web.post('/repo/{repo}/commit/{commit}/remove',
-                         server._post_remove_handler),
-        aiohttp.web.static('/', static_dir)])
+    get_routes = (
+        aiohttp.web.get(path, handler) for path, handler in (
+            ('/', ui._get_root_handler),
+            ('/main.css', ui._get_style_handler),
+            ('/repo/{repo}', ui._get_repo_handler),
+            ('/repo/{repo}/commit/{commit}', ui._get_commit_handler)))
+    post_routes = (
+        aiohttp.web.post(path, handler) for path, handler in (
+            ('/repo/{repo}/webhook', ui._post_webhook_handler),
+            ('/repo/{repo}/commit/{commit}/rerun', ui._post_rerun_handler),
+            ('/repo/{repo}/commit/{commit}/remove', ui._post_remove_handler)))
+    app.add_routes([*get_routes, *post_routes])
 
     runner = aiohttp.web.AppRunner(app)
     await runner.setup()
@@ -58,19 +59,139 @@ class UI(aio.Resource):
         return self._async_group
 
     async def _get_root_handler(self, request):
-        pass
+        repos = self._server.get_repos()
+        commits = await self._server.get_commits(None)
+        body = (f'{_generate_repos(repos)}\n'
+                f'{_generate_commits(commits)}')
+        return _create_html_response('hatter', body)
+
+    async def _get_style_handler(self, request):
+        return aiohttp.web.Response(content_type='text/css',
+                                    text=_main_css)
 
     async def _get_repo_handler(self, request):
-        pass
+        repo = request.match_info['repo']
+        commits = await self._server.get_commits(repo)
+        body = _generate_commits(commits)
+        return _create_html_response(f'hatter - {repo}', body)
 
     async def _get_commit_handler(self, request):
-        pass
+        repo = request.match_info['repo']
+        commit_hash = request.match_info['commit']
+        commit = await self._server.get_commit(repo, commit_hash)
+        body = _generate_commit(commit)
+        return _create_html_response(f'hatter - {repo}/{commit_hash}', body)
 
     async def _post_webhook_handler(self, request):
-        pass
+        repo = request.match_info['repo']
+        self._server.sync_repo(repo)
+        return aiohttp.web.Response()
 
-    async def _post_run_handler(self, request):
-        pass
+    async def _post_rerun_handler(self, request):
+        repo = request.match_info['repo']
+        commit_hash = request.match_info['commit']
+        await self._server.rerun_commit(repo, commit_hash)
+        raise aiohttp.web.HTTPFound(f'/repo/{repo}/commit/{commit_hash}')
 
     async def _post_remove_handler(self, request):
-        pass
+        repo = request.match_info['repo']
+        commit_hash = request.match_info['commit']
+        await self._server.remove_commit(repo, commit_hash)
+        raise aiohttp.web.HTTPFound(f'/repo/{repo}')
+
+
+def _create_html_response(title, body):
+    text = _html_template.format(title=title,
+                                 body=body)
+    return aiohttp.web.Response(content_type='text/html',
+                                text=text)
+
+
+def _generate_repos(repos):
+    items = '\n'.join(f'<li><a href="/repo/{repo}">{repo}</a></li>'
+                      for repo in repos)
+    return (f'<div class="repos">\n'
+            f'<ul>\n'
+            f'{items}\n'
+            f'</ul>\n'
+            f'</div>')
+
+
+def _generate_commits(commits):
+    thead = ('<tr>\n'
+             '<th class="col-change">Change</th>\n'
+             '<th class="col-repo">Repo</th>\n'
+             '<th class="col-hash">Commit</th>\n'
+             '<th class="col-status">Status</th>\n'
+             '</tr>')
+
+    tbody = '\n'.join(
+        (f'<tr>\n'
+         f'<td class="col-change">{_format_time(commit.change)}</td>\n'
+         f'<td class="col-repo"><a href="/repo/{commit.repo}">{commit.repo}</a></td>\n'  # NOQA
+         f'<td class="col-hash"><a href="/repo/{commit.repo}/commit/{commit.hash}">{commit.hash}</a></td>\n'  # NOQA
+         f'<td class="col-status">{commit.status.name}</td>\n'
+         f'</tr>')
+        for commit in commits)
+
+    return (f'<div class="commits">\n'
+            f'<table>\n'
+            f'<thead>\n'
+            f'{thead}\n'
+            f'</thead>\n'
+            f'<tbody>\n'
+            f'{tbody}\n'
+            f'</tbody>\n'
+            f'</table>\n'
+            f'</div>')
+
+
+def _generate_commit(commit):
+    buttons = '\n'.join(
+        (f'<form method="post" action="{action}">\n'
+         f'<input type="submit" value="{value}">\n'
+         f'</form>')
+        for value, action in (
+            ('Rerun', f'/repo/{commit.repo}/commit/{commit.hash}/rerun'),
+            ('Remove', f'/repo/{commit.repo}/commit/{commit.hash}/remove')))
+
+    return (f'<div class="commit">\n'
+            f'<label>Repo:</label><div><a href="/repo/{commit.repo}">{commit.repo}</a></div>\n'  # NOQA
+            f'<label>Commit:</label><div>{commit.hash}</div>\n'
+            f'<label>Change:</label><div>{_format_time(commit.change)}</div>\n'
+            f'<label>Status:</label><div>{commit.status.name}</div>\n'
+            f'<label>Output:</label><pre>{commit.output}</pre>\n'
+            f'<label></label><div>{buttons}</div>\n'
+            f'</div>')
+
+
+def _format_time(t):
+    return datetime.datetime.fromtimestamp(t).strftime("%Y-%m-%d %H:%M:%S")
+
+
+_html_template = r"""<!DOCTYPE html>
+<html>
+<head>
+<meta charset="UTF-8">
+<title>{title}</title>
+<link href="/main.css" rel="stylesheet">
+</head>
+<body>
+{body}
+</body>
+</html>
+"""
+
+_main_css = r"""
+.repos {
+
+}
+
+.commits {
+
+}
+
+.commit {
+
+}
+"""