# -*- coding: utf-8 -*-
# Copyright (C) 2013-2020 by The Linux Foundation and contributors
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
import stat
import sys

import grokmirror
import logging
import requests
import time
import gzip
import json
import fnmatch
import shutil
import tempfile
import signal
import shlex

import calendar
import uuid

import multiprocessing as mp
import queue

from socketserver import UnixStreamServer, StreamRequestHandler, ThreadingMixIn

# default basic logger. We override it later.
logger = logging.getLogger(__name__)


class SignalHandler:

    def __init__(self, config, sw, dws, pws, done):
        self.config = config
        self.sw = sw
        self.dws = dws
        self.pws = pws
        self.done = done
        self.killed = False

    def _handler(self, signum, frame):
        self.killed = True
        logger.debug('Received signum=%s, frame=%s', signum, frame)
        # if self.sw:
        #    self.sw.terminate()
        #    self.sw.join()

        # for dw in self.dws:
        #    if dw and dw.is_alive():
        #        dw.terminate()
        #        dw.join()

        # for pw in self.pws:
        #    if pw and pw.is_alive():
        #        pw.terminate()
        #        pw.join()

        if len(self.done):
            update_manifest(self.config, self.done)

        logger.info('Exiting on signal %s', signum)
        sys.exit(0)

    def __enter__(self):
        self.old_sigint = signal.signal(signal.SIGINT, self._handler)
        self.old_sigterm = signal.signal(signal.SIGTERM, self._handler)

    def __exit__(self, sigtype, value, traceback):
        if self.killed:
            sys.exit(0)
        signal.signal(signal.SIGINT, self.old_sigint)
        signal.signal(signal.SIGTERM, self.old_sigterm)


class Handler(StreamRequestHandler):

    def handle(self):
        config = self.server.config
        manifile = config['core'].get('manifest')
        while True:
            # noinspection PyBroadException
            try:
                gitdir = self.rfile.readline().strip().decode()
                # Do we know anything about this path?
                manifest = grokmirror.read_manifest(manifile)
                if gitdir in manifest:
                    logger.info(' listener: %s', gitdir)
                    repoinfo = manifest[gitdir]
                    # Set fingerprint to None to force a run
                    repoinfo['fingerprint'] = None
                    repoinfo['modified'] = int(time.time())
                    self.server.q_mani.put((gitdir, repoinfo, 'pull'))
                elif gitdir:
                    logger.info(' listener: %s (not known, ignored)', gitdir)
                    return
                else:
                    return
            except:
                return


class ThreadedUnixStreamServer(ThreadingMixIn, UnixStreamServer):
    pass


def build_optimal_forkgroups(l_manifest, r_manifest, toplevel, obstdir):
    r_forkgroups = dict()
    for gitdir in set(r_manifest.keys()):
        fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
        # our forkgroup info wins, because our own grok-fcsk may have found better siblings
        # unless we're cloning, in which case we have nothing to go by except remote info
        if gitdir in l_manifest:
            reference = l_manifest[gitdir].get('reference', None)
            forkgroup = l_manifest[gitdir].get('forkgroup', None)
            if reference is not None:
                r_manifest[gitdir]['reference'] = reference
            if forkgroup is not None:
                r_manifest[gitdir]['forkgroup'] = forkgroup
        else:
            reference = r_manifest[gitdir].get('reference', None)
            forkgroup = r_manifest[gitdir].get('forkgroup', None)

        if reference and not forkgroup:
            # probably a grokmirror-1.x manifest
            r_fullpath = os.path.join(toplevel, reference.lstrip('/'))
            for fg, fps in r_forkgroups.items():
                if r_fullpath in fps:
                    forkgroup = fg
                    break
            if not forkgroup:
                # I guess we get to make a new one!
                forkgroup = str(uuid.uuid4())
                r_forkgroups[forkgroup] = {r_fullpath}

        if forkgroup is not None:
            if forkgroup not in r_forkgroups:
                r_forkgroups[forkgroup] = set()
            r_forkgroups[forkgroup].add(fullpath)

    # Compare their forkgroups and my forkgroups in case we have a more optimal strategy
    forkgroups = grokmirror.get_forkgroups(obstdir, toplevel)
    for r_fg, r_siblings in r_forkgroups.items():
        # if we have an intersection between their forkgroups and our forkgroups, then we use ours
        found = False
        for l_fg, l_siblings in forkgroups.items():
            if l_siblings == r_siblings:
                # No changes there
                continue
            if len(l_siblings.intersection(r_siblings)):
                l_siblings.update(r_siblings)
                found = True
                break
        if not found:
            # We don't have any matches in existing repos, so make a new forkgroup
            forkgroups[r_fg] = r_siblings

    return forkgroups


def spa_worker(config, q_spa, pauseonload):
    toplevel = os.path.realpath(config['core'].get('toplevel'))
    cpus = mp.cpu_count()
    saidpaused = False
    while True:
        if pauseonload:
            load = os.getloadavg()
            if load[0] > cpus:
                if not saidpaused:
                    logger.info('      spa: paused (system load), %s waiting', q_spa.qsize())
                    saidpaused = True
                time.sleep(5)
                continue
            saidpaused = False

        try:
            (gitdir, actions) = q_spa.get(timeout=1)
        except queue.Empty:
            sys.exit(0)

        logger.debug('spa_worker: gitdir=%s, actions=%s', gitdir, actions)
        fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
        try:
            grokmirror.lock_repo(fullpath, nonblocking=True)
        except IOError:
            # We'll get it during grok-fsck
            continue

        if not q_spa.empty():
            logger.info('      spa: 1 active, %s waiting', q_spa.qsize())
        else:
            logger.info('      spa: 1 active')

        done = list()
        for action in actions:
            if action in done:
                continue
            done.append(action)
            if action == 'objstore':
                altrepo = grokmirror.get_altrepo(fullpath)
                # Should we use plumbing for this?
                use_plumbing = config['core'].getboolean('objstore_uses_plumbing', False)
                grokmirror.fetch_objstore_repo(altrepo, fullpath, use_plumbing=use_plumbing)

            elif action == 'repack':
                logger.debug('quick-repacking %s', fullpath)
                args = ['repack', '-Adlq']
                ecode, out, err = grokmirror.run_git_command(fullpath, args)
                if ecode > 0:
                    logger.debug('Could not repack %s', fullpath)

            elif action == 'packrefs':
                args = ['pack-refs']
                ecode, out, err = grokmirror.run_git_command(fullpath, args)
                if ecode > 0:
                    logger.debug('Could not pack-refs %s', fullpath)

            elif action == 'packrefs-all':
                args = ['pack-refs', '--all']
                ecode, out, err = grokmirror.run_git_command(fullpath, args)
                if ecode > 0:
                    logger.debug('Could not pack-refs %s', fullpath)

        grokmirror.unlock_repo(fullpath)
        logger.info('      spa: %s (done: %s)', gitdir, ', '.join(done))


def objstore_repo_preload(config, obstrepo):
    purl = config['remote'].get('preload_bundle_url')
    if not purl:
        return
    bname = os.path.basename(obstrepo)[:-4]
    obstdir = os.path.realpath(config['core'].get('objstore'))
    burl = '%s/%s.bundle' % (purl.rstrip('/'), bname)
    bfile = os.path.join(obstdir, '%s.bundle' % bname)
    sess = grokmirror.get_requests_session()
    resp = sess.get(burl, stream=True)
    resp.raise_for_status()
    logger.info(' objstore: getting preload bundle for %s', bname)
    with open(bfile, 'wb') as fh:
        for chunk in resp.iter_content(chunk_size=8192):
            fh.write(chunk)
    resp.close()

    # Now we clone from it into the objstore repo
    ecode, out, err = grokmirror.run_git_command(obstrepo, ['remote', 'add', '--mirror=fetch', '_preload', bfile])
    if ecode == 0:
        logger.info(' objstore: preloading %s from the bundle', bname)
        args = ['remote', 'update', '_preload']
        ecode, out, err = grokmirror.run_git_command(obstrepo, args)
        if ecode > 0:
            logger.info(' objstore: not able to preload, will clone repo-by-repo')
        else:
            # now pack refs and generate a commit graph
            grokmirror.run_git_command(obstrepo, ['pack-refs', '--all'])
            if grokmirror.git_newer_than('2.18.0'):
                grokmirror.run_git_command(obstrepo, ['commit-graph', 'write'])
            logger.info(' objstore: successful preload')
    # Regardless of what happened, we remove _preload and the bundle, then move on
    grokmirror.run_git_command(obstrepo, ['remote', 'rm', '_preload'])
    os.unlink(bfile)


def pull_worker(config, q_pull, q_spa, q_done):
    toplevel = os.path.realpath(config['core'].get('toplevel'))
    obstdir = os.path.realpath(config['core'].get('objstore'))
    maxretries = config['pull'].getint('retries', 3)
    site = config['remote'].get('site')
    remotename = config['pull'].get('remotename', '_grokmirror')
    # Should we use plumbing for objstore operations?
    objstore_uses_plumbing = config['core'].getboolean('objstore_uses_plumbing', False)

    while True:
        try:
            (gitdir, repoinfo, action, q_action) = q_pull.get(timeout=1)
        except queue.Empty:
            sys.exit(0)

        logger.debug('pull_worker: gitdir=%s, action=%s', gitdir, action)
        fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
        success = True
        spa_actions = list()

        try:
            grokmirror.lock_repo(fullpath, nonblocking=True)
        except IOError:
            # Take a quick nap and put it back into queue
            logger.info('    defer: %s (locked)', gitdir)
            time.sleep(5)
            q_pull.put((gitdir, repoinfo, action, q_action))
            continue

        altrepo = grokmirror.get_altrepo(fullpath)
        obstrepo = None
        if altrepo and grokmirror.is_obstrepo(altrepo, obstdir):
            obstrepo = altrepo

        if action == 'purge':
            # Is it a symlink?
            if os.path.islink(fullpath):
                logger.info('    purge: %s', gitdir)
                os.unlink(fullpath)
            else:
                # is anything using us for alternates?
                if grokmirror.is_alt_repo(toplevel, gitdir):
                    logger.debug('Not purging %s because it is used by other repos via alternates', fullpath)
                else:
                    logger.info('    purge: %s', gitdir)
                    shutil.rmtree(fullpath)

        if action == 'fix_remotes':
            logger.info(' reorigin: %s', gitdir)
            success = fix_remotes(toplevel, gitdir, site, config)
            if success:
                action = 'fix_params'
            else:
                success = False

        if action == 'fix_params':
            logger.info(' reconfig: %s', gitdir)
            set_repo_params(fullpath, repoinfo)
            action = 'pull'

        if action == 'reclone':
            logger.info('  reclone: %s', gitdir)
            try:
                altrepo = grokmirror.get_altrepo(fullpath)
                shutil.move(fullpath, '%s.reclone' % fullpath)
                shutil.rmtree('%s.reclone' % fullpath)
                grokmirror.setup_bare_repo(fullpath)
                fix_remotes(toplevel, gitdir, site, config)
                set_repo_params(fullpath, repoinfo)
                if altrepo:
                    grokmirror.set_altrepo(fullpath, altrepo)
                action = 'pull'
            except (PermissionError, IOError) as ex:
                logger.critical('Unable to remove %s: %s', fullpath, str(ex))
                success = False

        if action in ('pull', 'objstore_migrate'):
            r_fp = repoinfo.get('fingerprint')
            my_fp = grokmirror.get_repo_fingerprint(toplevel, gitdir, force=True)
            if obstrepo:
                o_obj_info = grokmirror.get_repo_obj_info(obstrepo)
                if o_obj_info.get('count') == '0' and o_obj_info.get('in-pack') == '0' and not my_fp:
                    # Try to preload the objstore repo directly
                    try:
                        objstore_repo_preload(config, obstrepo)
                    except: # noqa
                        logger.info(' objstore: not able to preload, will clone repo-by-repo')
                        # Make sure we don't leave .bundle files trailing around
                        bfile = obstrepo[:-4] + '.bundle'
                        if os.path.exists(bfile):
                            os.unlink(bfile)
                        pass

            if r_fp != my_fp:
                # Make sure we have the remote set up
                if action == 'pull' and remotename not in grokmirror.list_repo_remotes(fullpath):
                    logger.info(' reorigin: %s', gitdir)
                    fix_remotes(toplevel, gitdir, site, config)
                logger.info('    fetch: %s', gitdir)
                retries = 1
                while True:
                    success = pull_repo(fullpath, remotename)
                    if success:
                        break
                    retries += 1
                    if retries > maxretries:
                        break
                    logger.info('  refetch: %s (try #%s)', gitdir, retries)

                if success:
                    run_post_update_hook(toplevel, gitdir, config['pull'].get('post_update_hook', ''))
                    post_pull_fp = grokmirror.get_repo_fingerprint(toplevel, gitdir, force=True)
                    repoinfo['fingerprint'] = post_pull_fp
                    altrepo = grokmirror.get_altrepo(fullpath)
                    if post_pull_fp != my_fp:
                        grokmirror.set_repo_fingerprint(toplevel, gitdir, fingerprint=post_pull_fp)
                        if altrepo and grokmirror.is_obstrepo(altrepo, obstdir) and not repoinfo.get('private'):
                            # do we have any objects in the objstore repo?
                            o_obj_info = grokmirror.get_repo_obj_info(altrepo)
                            if o_obj_info.get('count') == '0' and o_obj_info.get('in-pack') == '0':
                                # We fetch right now, as other repos may be waiting on these objects
                                logger.info(' objstore: %s', gitdir)
                                grokmirror.fetch_objstore_repo(altrepo, fullpath, use_plumbing=objstore_uses_plumbing)
                                if not objstore_uses_plumbing:
                                    spa_actions.append('repack')
                            else:
                                # We lazy-fetch in the spa
                                spa_actions.append('objstore')
                                if my_fp is None and not objstore_uses_plumbing:
                                    # Initial clone, trigger a repack after objstore
                                    spa_actions.append('repack')

                        if my_fp is None:
                            # This was the initial clone, so pack all refs
                            spa_actions.append('packrefs-all')

                        if not grokmirror.is_precious(fullpath):
                            # See if doing a quick repack would be beneficial
                            obj_info = grokmirror.get_repo_obj_info(fullpath)
                            if grokmirror.get_repack_level(obj_info):
                                # We only do quick repacks, so we don't care about precise level
                                spa_actions.append('repack')
                                spa_actions.append('packrefs')

                    modified = repoinfo.get('modified')
                    if modified is not None:
                        set_agefile(toplevel, gitdir, modified)
            else:
                logger.debug('FP match, not pulling %s', gitdir)

        if action == 'objstore_migrate':
            spa_actions.append('objstore')
            spa_actions.append('repack')

        grokmirror.unlock_repo(fullpath)

        symlinks = repoinfo.get('symlinks')
        if os.path.exists(fullpath) and symlinks:
            for symlink in symlinks:
                target = os.path.join(toplevel, symlink.lstrip('/'))

                if os.path.islink(target):
                    # are you pointing to where we need you?
                    if os.path.realpath(target) != fullpath:
                        # Remove symlink and recreate below
                        logger.debug('Removed existing wrong symlink %s', target)
                        os.unlink(target)
                elif os.path.exists(target):
                    logger.warning('Deleted repo %s, because it is now a symlink to %s' % (target, fullpath))
                    shutil.rmtree(target)

                # Here we re-check if we still need to do anything
                if not os.path.exists(target):
                    logger.info('  symlink: %s -> %s', symlink, gitdir)
                    # Make sure the leading dirs are in place
                    if not os.path.exists(os.path.dirname(target)):
                        os.makedirs(os.path.dirname(target))
                    os.symlink(fullpath, target)

        q_done.put((gitdir, repoinfo, q_action, success))
        if spa_actions:
            q_spa.put((gitdir, spa_actions))


def cull_manifest(manifest, config):
    includes = config['pull'].get('include', '*').split('\n')
    excludes = config['pull'].get('exclude', '').split('\n')

    culled = dict()

    for gitdir, repoinfo in manifest.items():
        if not repoinfo.get('fingerprint'):
            logger.critical('Repo without fingerprint info (skipped): %s', gitdir)
            continue
        # does it fall under include?
        for include in includes:
            if fnmatch.fnmatch(gitdir, include):
                # Yes, but does it fall under excludes?
                excluded = False
                for exclude in excludes:
                    if fnmatch.fnmatch(gitdir, exclude):
                        excluded = True
                        break
                if excluded:
                    continue

                culled[gitdir] = manifest[gitdir]

    return culled


def fix_remotes(toplevel, gitdir, site, config):
    remotename = config['pull'].get('remotename', '_grokmirror')
    fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
    # Set our remote
    if remotename in grokmirror.list_repo_remotes(fullpath):
        logger.debug('\tremoving remote: %s', remotename)
        ecode, out, err = grokmirror.run_git_command(fullpath, ['remote', 'remove', remotename])
        if ecode > 0:
            logger.critical('FATAL: Could not remove remote %s from %s', remotename, fullpath)
            return False

    # set my remote URL
    url = os.path.join(site, gitdir.lstrip('/'))
    ecode, out, err = grokmirror.run_git_command(fullpath, ['remote', 'add', '--mirror=fetch', remotename, url])
    if ecode > 0:
        logger.critical('FATAL: Could not set %s to %s in %s', remotename, url, fullpath)
        return False

    ffonly = False
    for globpatt in set([x.strip() for x in config['pull'].get('ffonly', '').split('\n')]):
        if fnmatch.fnmatch(gitdir, globpatt):
            ffonly = True
            break
    if ffonly:
        grokmirror.set_git_config(fullpath, 'remote.{}.fetch'.format(remotename), 'refs/*:refs/*')
        logger.debug('\tset %s as %s (ff-only)', remotename, url)
    else:
        logger.debug('\tset %s as %s', remotename, url)
    return True


def set_repo_params(fullpath, repoinfo):
    owner = repoinfo.get('owner')
    description = repoinfo.get('description')
    head = repoinfo.get('head')
    if owner is None and description is None and head is None:
        # Let the default git values be there, then
        return

    if description is not None:
        descfile = os.path.join(fullpath, 'description')
        contents = None
        if os.path.exists(descfile):
            with open(descfile) as fh:
                contents = fh.read()
        if contents != description:
            logger.debug('Setting %s description to: %s', fullpath, description)
            with open(descfile, 'w') as fh:
                fh.write(description)

    if owner is not None:
        logger.debug('Setting %s owner to: %s', fullpath, owner)
        grokmirror.set_git_config(fullpath, 'gitweb.owner', owner)

    if head is not None:
        headfile = os.path.join(fullpath, 'HEAD')
        contents = None
        if os.path.exists(headfile):
            with open(headfile) as fh:
                contents = fh.read().rstrip()
        if contents != head:
            logger.debug('Setting %s HEAD to: %s', fullpath, head)
            with open(headfile, 'w') as fh:
                fh.write('{}\n'.format(head))


def set_agefile(toplevel, gitdir, last_modified):
    grokmirror.set_repo_timestamp(toplevel, gitdir, last_modified)

    # set agefile, which can be used by cgit to show idle times
    # cgit recommends it to be yyyy-mm-dd hh:mm:ss
    cgit_fmt = time.strftime('%F %T', time.localtime(last_modified))
    agefile = os.path.join(toplevel, gitdir.lstrip('/'), 'info/web/last-modified')
    if not os.path.exists(os.path.dirname(agefile)):
        os.makedirs(os.path.dirname(agefile))
    with open(agefile, 'wt') as fh:
        fh.write('%s\n' % cgit_fmt)
    logger.debug('Wrote "%s" into %s', cgit_fmt, agefile)


def run_post_update_hook(toplevel, gitdir, hookscripts):
    if not len(hookscripts):
        return

    for hookscript in hookscripts.split('\n'):
        hookscript = os.path.expanduser(hookscript.strip())
        sp = shlex.shlex(hookscript, posix=True)
        sp.whitespace_split = True
        args = list(sp)

        logger.info('     hook: %s', ' '.join(args))
        if not os.access(args[0], os.X_OK):
            logger.warning('post_update_hook %s is not executable', hookscript)
            continue

        fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
        args.append(fullpath)
        logger.debug('Running: %s', ' '.join(args))
        ecode, output, error = grokmirror.run_shell_command(args)

        if error:
            # Put hook stderror into warning
            logger.warning('Hook Stderr (%s): %s', gitdir, error)
        if output:
            # Put hook stdout into info
            logger.info('Hook Stdout (%s): %s', gitdir, output)


def pull_repo(fullpath, remotename):
    args = ['remote', 'update', remotename, '--prune']

    retcode, output, error = grokmirror.run_git_command(fullpath, args)

    success = False
    if retcode == 0:
        success = True

    if error:
        # Put things we recognize into debug
        debug = list()
        warn = list()
        for line in error.split('\n'):
            if line.find('From ') == 0:
                debug.append(line)
            elif line.find('-> ') > 0:
                debug.append(line)
            elif line.find('remote: warning:') == 0:
                debug.append(line)
            elif line.find('ControlSocket') >= 0:
                debug.append(line)
            elif not success:
                warn.append(line)
            else:
                debug.append(line)
        if debug:
            logger.debug('Stderr (%s): %s', fullpath, '\n'.join(debug))
        if warn:
            logger.warning('Stderr (%s): %s', fullpath, '\n'.join(warn))

    return success


def write_projects_list(config, manifest):
    plpath = config['pull'].get('projectslist', '')
    if not plpath:
        return

    trimtop = config['pull'].get('projectslist_trimtop', '')
    add_symlinks = config['pull'].getboolean('projectslist_symlinks', False)

    (dirname, basename) = os.path.split(plpath)
    (fd, tmpfile) = tempfile.mkstemp(prefix=basename, dir=dirname)

    try:
        fh = os.fdopen(fd, 'wb', 0)
        for gitdir in manifest:
            if trimtop and gitdir.startswith(trimtop):
                pgitdir = gitdir[len(trimtop):]
            else:
                pgitdir = gitdir

            # Always remove leading slash, otherwise cgit breaks
            pgitdir = pgitdir.lstrip('/')
            fh.write('{}\n'.format(pgitdir).encode())

            if add_symlinks and 'symlinks' in manifest[gitdir]:
                # Do the same for symlinks
                # XXX: Should make this configurable, perhaps
                for symlink in manifest[gitdir]['symlinks']:
                    if trimtop and symlink.startswith(trimtop):
                        symlink = symlink[len(trimtop):]

                    symlink = symlink.lstrip('/')
                    fh.write('{}\n'.format(symlink).encode())

        os.fsync(fd)
        fh.close()
        # set mode to current umask
        curmask = os.umask(0)
        os.chmod(tmpfile, 0o0666 ^ curmask)
        os.umask(curmask)
        shutil.move(tmpfile, plpath)

    finally:
        # If something failed, don't leave tempfiles trailing around
        if os.path.exists(tmpfile):
            os.unlink(tmpfile)

    logger.info(' projlist: wrote %s', plpath)


def fill_todo_from_manifest(config, q_mani, nomtime=False, forcepurge=False):
    # l_ = local, r_ = remote
    l_mani_path = config['core'].get('manifest')
    r_mani_cmd = config['remote'].get('manifest_command')

    if r_mani_cmd:
        if not os.access(r_mani_cmd, os.X_OK):
            logger.critical('Remote manifest command is not executable: %s', r_mani_cmd)
            sys.exit(1)
        logger.info(' manifest: executing %s', r_mani_cmd)
        cmdargs = [r_mani_cmd]
        if nomtime:
            cmdargs += ['--force']
        (ecode, output, error) = grokmirror.run_shell_command(cmdargs)
        if ecode == 0:
            try:
                r_manifest = json.loads(output)
            except json.JSONDecodeError as ex:
                logger.warning('Failed to parse output from %s', r_mani_cmd)
                logger.warning('Error was: %s', ex)
                raise IOError('Failed to parse output from %s (%s)' % (r_mani_cmd, ex))
        elif ecode == 127:
            logger.info(' manifest: unchanged')
            return
        elif ecode == 1:
            logger.warning('Executing %s failed, exiting', r_mani_cmd, ecode)
            raise IOError('Failed executing %s' % r_mani_cmd)
        else:
            # Non-fatal errors for all other exit codes
            logger.warning(' manifest: executing %s returned %s', r_mani_cmd, ecode)
            return

        if not len(r_manifest):
            logger.warning(' manifest: empty, ignoring')
            raise IOError('Empty manifest returned by %s' % r_mani_cmd)

    else:
        r_mani_status_path = os.path.join(os.path.dirname(l_mani_path), '.%s.remote' % os.path.basename(l_mani_path))
        try:
            with open(r_mani_status_path, 'r') as fh:
                r_mani_status = json.loads(fh.read())
        except (IOError, json.JSONDecodeError):
            logger.debug('Could not read %s', r_mani_status_path)
            r_mani_status = dict()
        r_last_fetched = r_mani_status.get('last-fetched', 0)
        config_last_modified = r_mani_status.get('config-last-modified', 0)
        if config_last_modified != config.last_modified:
            nomtime = True
        r_mani_url = config['remote'].get('manifest')
        logger.info(' manifest: fetching %s', r_mani_url)
        if r_mani_url.find('file:///') == 0:
            r_mani_url = r_mani_url.replace('file://', '')
            if not os.path.exists(r_mani_url):
                logger.critical('Remote manifest not found in %s! Quitting!', r_mani_url)
                raise IOError('Remote manifest not found in %s' % r_mani_url)

            fstat = os.stat(r_mani_url)
            r_last_modified = fstat[8]
            if r_last_fetched:
                logger.debug('mtime on %s is: %s', r_mani_url, fstat[8])
                if not nomtime and r_last_modified <= r_last_fetched:
                    logger.info(' manifest: unchanged')
                    return

            logger.info('Reading new manifest from %s', r_mani_url)
            r_manifest = grokmirror.read_manifest(r_mani_url)
            # Don't accept empty manifests -- that indicates something is wrong
            if not len(r_manifest):
                logger.warning('Remote manifest empty or unparseable! Quitting.')
                raise IOError('Empty manifest in %s' % r_mani_url)

        else:
            session = grokmirror.get_requests_session()

            # Find out if we need to run at all first
            headers = dict()
            if r_last_fetched and not nomtime:
                last_modified_h = time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime(r_last_fetched))
                logger.debug('Our last-modified is: %s', last_modified_h)
                headers['If-Modified-Since'] = last_modified_h

            try:
                # 30 seconds to connect, 5 minutes between reads
                res = session.get(r_mani_url, headers=headers, timeout=(30, 300))
            except requests.exceptions.RequestException as ex:
                logger.warning('Could not fetch %s', r_mani_url)
                logger.warning('Server returned: %s', ex)
                raise IOError('Remote server returned an error: %s' % ex)

            if res.status_code == 304:
                # No change to the manifest, nothing to do
                logger.info(' manifest: unchanged')
                return

            if res.status_code > 200:
                logger.warning('Could not fetch %s', r_mani_url)
                logger.warning('Server returned status: %s', res.status_code)
                raise IOError('Remote server returned an error: %s' % res.status_code)

            r_last_modified = res.headers['Last-Modified']
            r_last_modified = time.strptime(r_last_modified, '%a, %d %b %Y %H:%M:%S %Z')
            r_last_modified = calendar.timegm(r_last_modified)

            # We don't use read_manifest for the remote manifest, as it can be
            # anything, really. For now, blindly open it with gzipfile if it ends
            # with .gz. XXX: some http servers will auto-deflate such files.
            try:
                if r_mani_url.rfind('.gz') > 0:
                    import io
                    fh = gzip.GzipFile(fileobj=io.BytesIO(res.content))
                    jdata = fh.read().decode()
                else:
                    jdata = res.content

                res.close()
                # Don't hold session open, since we don't refetch manifest very frequently
                session.close()
                r_manifest = json.loads(jdata)

            except Exception as ex:
                logger.warning('Failed to parse %s', r_mani_url)
                logger.warning('Error was: %s', ex)
                raise IOError('Failed to parse %s (%s)' % (r_mani_url, ex))

        # Record for the next run
        with open(r_mani_status_path, 'w') as fh:
            r_mani_status = {
                'source': r_mani_url,
                'last-fetched': r_last_modified,
                'config-last-modified': config.last_modified,
            }
            json.dump(r_mani_status, fh)

    l_manifest = grokmirror.read_manifest(l_mani_path)
    r_culled = cull_manifest(r_manifest, config)
    logger.info(' manifest: %s relevant entries', len(r_culled))

    toplevel = os.path.realpath(config['core'].get('toplevel'))

    obstdir = os.path.realpath(config['core'].get('objstore'))
    forkgroups = build_optimal_forkgroups(l_manifest, r_culled, toplevel, obstdir)
    privmasks = set([x.strip() for x in config['core'].get('private', '').split('\n')])

    # populate private/forkgroup info in r_culled
    for forkgroup, siblings in forkgroups.items():
        for s_fullpath in siblings:
            s_gitdir = '/' + os.path.relpath(s_fullpath, toplevel)

            is_private = False
            for privmask in privmasks:
                # Does this repo match privrepo
                if fnmatch.fnmatch(s_gitdir, privmask):
                    is_private = True
                    break
            if s_gitdir in r_culled:
                r_culled[s_gitdir]['forkgroup'] = forkgroup
                r_culled[s_gitdir]['private'] = is_private

    seen = set()
    to_migrate = set()
    # Used to track symlinks so we can properly avoid purging them
    all_symlinks = set()

    for gitdir, repoinfo in r_culled.items():
        symlinks = repoinfo.get('symlinks')
        if symlinks and isinstance(symlinks, list):
            all_symlinks.update(set(symlinks))

        if gitdir in seen:
            continue
        seen.add(gitdir)
        fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
        forkgroup = repoinfo.get('forkgroup')

        # Is the directory in place?
        if os.path.exists(fullpath):
            # Did grok-fsck request to reclone it?
            rfile = os.path.join(fullpath, 'grokmirror.reclone')
            if os.path.exists(rfile):
                logger.debug('Reclone requested for %s:', gitdir)
                q_mani.put((gitdir, repoinfo, 'reclone'))
                with open(rfile, 'r') as rfh:
                    reason = rfh.read()
                    logger.debug('  %s', reason)
                continue

            if gitdir not in l_manifest:
                q_mani.put((gitdir, repoinfo, 'fix_remotes'))
                continue

            r_desc = r_culled[gitdir].get('description')
            r_owner = r_culled[gitdir].get('owner')
            r_head = r_culled[gitdir].get('head')

            l_desc = l_manifest[gitdir].get('description')
            l_owner = l_manifest[gitdir].get('owner')
            l_head = l_manifest[gitdir].get('head')

            if l_owner is None:
                l_owner = config['pull'].get('default_owner', 'Grokmirror')
            if r_owner is None:
                r_owner = config['pull'].get('default_owner', 'Grokmirror')

            if r_desc != l_desc or r_owner != l_owner or r_head != l_head:
                q_mani.put((gitdir, repoinfo, 'fix_params'))

            my_fingerprint = grokmirror.get_repo_fingerprint(toplevel, gitdir)
            if my_fingerprint != l_manifest[gitdir].get('fingerprint'):
                logger.debug('Fingerprint discrepancy, forcing a fetch')
                q_mani.put((gitdir, repoinfo, 'pull'))
                continue

            if my_fingerprint == r_culled[gitdir]['fingerprint']:
                logger.debug('Fingerprints match, skipping %s', gitdir)
                continue

            logger.debug('No fingerprint match, will pull %s', gitdir)
            q_mani.put((gitdir, repoinfo, 'pull'))
            continue

        if not forkgroup:
            # no-sibling repo
            q_mani.put((gitdir, repoinfo, 'init'))
            continue

        obstrepo = os.path.join(obstdir, '%s.git' % forkgroup)
        if os.path.isdir(obstrepo):
            # Init with an existing obstrepo, easy case
            q_mani.put((gitdir, repoinfo, 'init'))
            continue

        # Do we have any existing siblings that were cloned without obstrepo?
        # This would happen when an initial fork is created of an existing repo.
        found_existing = False
        public_siblings = set()
        for s_fullpath in forkgroups[forkgroup]:
            s_gitdir = '/' + os.path.relpath(s_fullpath, toplevel)
            if s_gitdir == gitdir:
                continue

            # can't simply rely on r_culled 'private' info, as this repo may only exist locally
            is_private = False
            for privmask in privmasks:
                # Does this repo match privrepo
                if fnmatch.fnmatch(s_gitdir, privmask):
                    is_private = True
                    break
            if is_private:
                # Can't use this sibling for anything, as it's private
                continue

            if os.path.isdir(s_fullpath):
                found_existing = True
                if s_gitdir not in to_migrate:
                    # Plan to migrate it to objstore
                    logger.debug('reusing existing %s as new obstrepo %s', s_gitdir, obstrepo)
                    s_repoinfo = grokmirror.get_repo_defs(toplevel, s_gitdir, usenow=True)
                    s_repoinfo['forkgroup'] = forkgroup
                    s_repoinfo['private'] = False
                    # Stick it into queue before the new clone
                    q_mani.put((s_gitdir, s_repoinfo, 'objstore_migrate'))
                    seen.add(s_gitdir)
                    to_migrate.add(s_gitdir)
                break
            if s_gitdir in r_culled:
                public_siblings.add(s_gitdir)

        if found_existing:
            q_mani.put((gitdir, repoinfo, 'init'))
            continue

        if repoinfo['private'] and len(public_siblings):
            # Clone public siblings first
            for s_gitdir in public_siblings:
                if s_gitdir not in seen:
                    q_mani.put((s_gitdir, r_culled[s_gitdir], 'init'))
                    seen.add(s_gitdir)
        # Finally, clone ourselves.
        q_mani.put((gitdir, repoinfo, 'init'))

    if config['pull'].getboolean('purge', False):
        nopurge = config['pull'].get('nopurge', '').split('\n')
        to_purge = set()
        found_repos = 0
        for founddir in grokmirror.find_all_gitdirs(toplevel, exclude_objstore=True):
            gitdir = '/' + os.path.relpath(founddir, toplevel)
            found_repos += 1

            if gitdir not in r_culled and gitdir not in all_symlinks:
                exclude = False
                for entry in nopurge:
                    if fnmatch.fnmatch(gitdir, entry):
                        exclude = True
                        break
                # Refuse to purge ffonly repos
                for globpatt in set([x.strip() for x in config['pull'].get('ffonly', '').split('\n')]):
                    if fnmatch.fnmatch(gitdir, globpatt):
                        # Woah, these are not supposed to be deleted, ever
                        logger.critical('Refusing to purge ffonly repo %s', gitdir)
                        exclude = True
                        break
                if not exclude:
                    to_purge.add(gitdir)

        if len(to_purge):
            # Purge-protection engage
            purge_limit = int(config['pull'].getint('purgeprotect', 5))
            if purge_limit < 1 or purge_limit > 99:
                logger.critical('Warning: "%s" is not valid for purgeprotect.', purge_limit)
                logger.critical('Please set to a number between 1 and 99.')
                logger.critical('Defaulting to purgeprotect=5.')
                purge_limit = 5

            purge_pc = int(len(to_purge) * 100 / found_repos)
            logger.debug('purgeprotect=%s', purge_limit)
            logger.debug('purge prercentage=%s', purge_pc)

            if not forcepurge and purge_pc >= purge_limit:
                logger.critical('Refusing to purge %s repos (%s%%)', len(to_purge), purge_pc)
                logger.critical('Set purgeprotect to a higher percentage, or override with --force-purge.')
            else:
                for gitdir in to_purge:
                    q_mani.put((gitdir, None, 'purge'))
        else:
            logger.debug('No repositories need purging')


def update_manifest(config, entries):
    manifile = config['core'].get('manifest')
    grokmirror.manifest_lock(manifile)
    manifest = grokmirror.read_manifest(manifile)
    changed = False
    while len(entries):
        gitdir, repoinfo, action, success = entries.pop()
        if not success:
            continue
        if action == 'purge':
            # Remove entry from manifest
            try:
                manifest.pop(gitdir)
                changed = True
            except KeyError:
                pass
            continue

        try:
            # does not belong in the manifest
            repoinfo.pop('private')
        except KeyError:
            pass
        for key, val in dict(repoinfo).items():
            # Clean up grok-2.0 null values
            if key in ('head', 'forkgroup') and val is None:
                repoinfo.pop(key)
        # Make sure 'reference' is present to prevent grok-1.x breakage
        if 'reference' not in repoinfo:
            repoinfo['reference'] = None
        manifest[gitdir] = repoinfo
        changed = True
    if changed:
        if 'manifest' in config:
            pretty = config['manifest'].getboolean('pretty', False)
        else:
            pretty = False
        grokmirror.write_manifest(manifile, manifest, pretty=pretty)
        logger.info(' manifest: wrote %s (%d entries)', manifile, len(manifest))
        # write out projects.list, if asked to
        write_projects_list(config, manifest)

    grokmirror.manifest_unlock(manifile)


def socket_worker(config, q_mani, sockfile):
    logger.info(' listener: listening on socket %s', sockfile)
    curmask = os.umask(0)
    with ThreadedUnixStreamServer(sockfile, Handler) as server:
        os.umask(curmask)
        # Stick some objects into the server
        server.q_mani = q_mani
        server.config = config
        server.serve_forever()


def showstats(q_todo, q_pull, q_spa, good, bad, pws, dws):
    stats = list()
    if good:
        stats.append('%s fetched' % good)
    if pws:
        stats.append('%s active' % len(pws))
    if not q_pull.empty():
        stats.append('%s queued' % q_pull.qsize())
    if not q_todo.empty():
        stats.append('%s waiting' % q_todo.qsize())
    if len(dws) or not q_spa.empty():
        stats.append('%s in spa' % (q_spa.qsize() + len(dws)))
    if bad:
        stats.append('%s errors' % bad)

    logger.info('      ---:  %s', ', '.join(stats))


def manifest_worker(config, q_mani, nomtime=False):
    starttime = int(time.time())
    fill_todo_from_manifest(config, q_mani, nomtime=nomtime)
    refresh = config['pull'].getint('refresh', 300)
    left = refresh - int(time.time() - starttime)
    if left > 0:
        logger.info(' manifest: sleeping %ss', left)


def pull_mirror(config, nomtime=False, forcepurge=False, runonce=False):
    toplevel = os.path.realpath(config['core'].get('toplevel'))
    obstdir = os.path.realpath(config['core'].get('objstore'))
    refresh = config['pull'].getint('refresh', 300)

    q_mani = mp.Queue()
    q_todo = mp.Queue()
    q_pull = mp.Queue()
    q_done = mp.Queue()
    q_spa = mp.Queue()

    sw = None
    sockfile = config['pull'].get('socket')
    if sockfile and not runonce:
        if os.path.exists(sockfile):
            mode = os.stat(sockfile).st_mode
            if stat.S_ISSOCK(mode):
                os.unlink(sockfile)
            else:
                raise IOError('File exists but is not a socket: %s' % sockfile)

        sw = mp.Process(target=socket_worker, args=(config, q_mani, sockfile))
        sw.daemon = True
        sw.start()

    pws = list()
    dws = list()
    mws = list()
    actions = set()
    # Run in the main thread if we have runonce
    if runonce:
        fill_todo_from_manifest(config, q_mani, nomtime=nomtime, forcepurge=forcepurge)
        if q_mani.empty():
            return 0
    else:
        # force nomtime to True the first time
        nomtime = True
    lastrun = 0

    pull_threads = config['pull'].getint('pull_threads', 0)
    if pull_threads < 1:
        # take half of available CPUs by default
        pull_threads = int(mp.cpu_count() / 2)

    busy = set()
    done = list()
    good = 0
    bad = 0
    loopmark = None
    with SignalHandler(config, sw, dws, pws, done):
        while True:
            for pw in pws:
                if pw and not pw.is_alive():
                    pws.remove(pw)
                    logger.info('   worker: terminated (%s remaining)', len(pws))
                    showstats(q_todo, q_pull, q_spa, good, bad, pws, dws)

            for dw in dws:
                if dw and not dw.is_alive():
                    dws.remove(dw)
                    showstats(q_todo, q_pull, q_spa, good, bad, pws, dws)

            for mw in mws:
                if mw and not mw.is_alive():
                    mws.remove(mw)

            if not q_spa.empty() and not len(dws):
                if runonce:
                    pauseonload = False
                else:
                    pauseonload = True
                dw = mp.Process(target=spa_worker, args=(config, q_spa, pauseonload))
                dw.daemon = True
                dw.start()
                dws.append(dw)

            if not q_pull.empty() and len(pws) < pull_threads:
                pw = mp.Process(target=pull_worker, args=(config, q_pull, q_spa, q_done))
                pw.daemon = True
                pw.start()
                pws.append(pw)
                logger.info('   worker: started (%s running)', len(pws))

            # Any new results?
            try:
                while True:
                    gitdir, repoinfo, q_action, success = q_done.get_nowait()
                    try:
                        actions.remove((gitdir, q_action))
                    except KeyError:
                        pass

                    forkgroup = repoinfo.get('forkgroup')
                    if forkgroup and forkgroup in busy:
                        busy.remove(forkgroup)
                    done.append((gitdir, repoinfo, q_action, success))
                    if success:
                        good += 1
                    else:
                        bad += 1
                    logger.info('     done: %s', gitdir)
                    showstats(q_todo, q_pull, q_spa, good, bad, pws, dws)
                    if len(done) >= 100:
                        # Write manifest every 100 repos
                        update_manifest(config, done)

            except queue.Empty:
                pass

            # Anything new in the manifest queue?
            try:
                new_updates = 0
                while True:
                    gitdir, repoinfo, action = q_mani.get_nowait()
                    if (gitdir, action) in actions:
                        logger.debug('already in the queue: %s, %s', gitdir, action)
                        continue
                    if action == 'pull' and (gitdir, 'init') in actions:
                        logger.debug('already in the queue as init: %s, %s', gitdir, action)
                        continue

                    actions.add((gitdir, action))
                    q_todo.put((gitdir, repoinfo, action))
                    new_updates += 1
                    logger.debug('queued: %s, %s', gitdir, action)

                if new_updates:
                    logger.info(' manifest: %s new updates', new_updates)

            except queue.Empty:
                pass

            if not runonce and not len(mws) and q_todo.empty() and q_pull.empty() and time.time() - lastrun >= refresh:
                if done:
                    update_manifest(config, done)
                mw = mp.Process(target=manifest_worker, args=(config, q_mani, nomtime))
                nomtime = False
                mw.daemon = True
                mw.start()
                mws.append(mw)
                lastrun = int(time.time())

            # Finally, deal with q_todo
            try:
                gitdir, repoinfo, q_action = q_todo.get_nowait()
                logger.debug('main_thread: got %s/%s from q_todo', gitdir, q_action)
            except queue.Empty:
                if q_mani.empty() and q_done.empty():
                    if not len(pws):
                        if done:
                            update_manifest(config, done)
                            if runonce:
                                # Wait till spa is done
                                while True:
                                    if q_spa.empty():
                                        for dw in dws:
                                            dw.join()
                                        return 0
                                    time.sleep(1)
                    # Don't run a hot loop waiting on results
                    time.sleep(5)
                continue

            if repoinfo is None:
                repoinfo = dict()

            fullpath = os.path.join(toplevel, gitdir.lstrip('/'))
            forkgroup = repoinfo.get('forkgroup')
            if gitdir in busy or (forkgroup is not None and forkgroup in busy):
                # Stick it back into the queue
                q_todo.put((gitdir, repoinfo, q_action))
                if loopmark is None:
                    loopmark = gitdir
                elif loopmark == gitdir:
                    # We've looped around all waiting repos, so back off and don't run
                    # a hot waiting loop.
                    time.sleep(5)
                continue

            if gitdir == loopmark:
                loopmark = None

            if q_action == 'objstore_migrate':
                # Add forkgroup to busy, so we don't run any pulls until it's done
                busy.add(repoinfo['forkgroup'])
                obstrepo = grokmirror.setup_objstore_repo(obstdir, name=forkgroup)
                grokmirror.add_repo_to_objstore(obstrepo, fullpath)
                grokmirror.set_altrepo(fullpath, obstrepo)

            if q_action != 'init':
                # Easy actions that don't require priority logic
                q_pull.put((gitdir, repoinfo, q_action, q_action))
                continue

            try:
                grokmirror.lock_repo(fullpath, nonblocking=True)
            except IOError:
                if not runonce:
                    q_todo.put((gitdir, repoinfo, q_action))
                continue

            if not grokmirror.setup_bare_repo(fullpath):
                logger.critical('Unable to bare-init %s', fullpath)
                q_done.put((gitdir, repoinfo, q_action, False))
                continue

            fix_remotes(toplevel, gitdir, config['remote'].get('site'), config)
            set_repo_params(fullpath, repoinfo)
            grokmirror.unlock_repo(fullpath)

            forkgroup = repoinfo.get('forkgroup')
            if not forkgroup:
                logger.debug('no-sibling clone: %s', gitdir)
                q_pull.put((gitdir, repoinfo, 'pull', q_action))
                continue

            obstrepo = os.path.join(obstdir, '%s.git' % forkgroup)
            if os.path.isdir(obstrepo):
                logger.debug('clone %s with existing obstrepo %s', gitdir, obstrepo)
                grokmirror.set_altrepo(fullpath, obstrepo)
                if not repoinfo['private']:
                    grokmirror.add_repo_to_objstore(obstrepo, fullpath)
                q_pull.put((gitdir, repoinfo, 'pull', q_action))
                continue

            # Set up a new obstrepo and make sure it's not used until the initial
            # pull is done
            logger.debug('cloning %s with new obstrepo %s', gitdir, obstrepo)
            busy.add(forkgroup)
            obstrepo = grokmirror.setup_objstore_repo(obstdir, name=forkgroup)
            grokmirror.set_altrepo(fullpath, obstrepo)
            if not repoinfo['private']:
                grokmirror.add_repo_to_objstore(obstrepo, fullpath)
            q_pull.put((gitdir, repoinfo, 'pull', q_action))

    return 0


def parse_args():
    import argparse
    # noinspection PyTypeChecker
    op = argparse.ArgumentParser(prog='grok-pull',
                                 description='Create or update a git repository collection mirror',
                                 formatter_class=argparse.ArgumentDefaultsHelpFormatter)

    op.add_argument('-v', '--verbose', dest='verbose', action='store_true',
                    default=False,
                    help='Be verbose and tell us what you are doing')
    op.add_argument('-n', '--no-mtime-check', dest='nomtime',
                    action='store_true', default=False,
                    help='Run without checking manifest mtime')
    op.add_argument('-p', '--purge', dest='purge',
                    action='store_true', default=False,
                    help='Remove any git trees that are no longer in manifest')
    op.add_argument('--force-purge', dest='forcepurge',
                    action='store_true', default=False,
                    help='Force purge despite significant repo deletions')
    op.add_argument('-o', '--continuous', dest='runonce',
                    action='store_false', default=True,
                    help='Run continuously (no effect if refresh is not set in config)')
    op.add_argument('-c', '--config', dest='config',
                    required=True,
                    help='Location of the configuration file')
    op.add_argument('--version', action='version', version=grokmirror.VERSION)

    return op.parse_args()


def grok_pull(cfgfile, verbose=False, nomtime=False, purge=False, forcepurge=False, runonce=False):
    global logger

    config = grokmirror.load_config_file(cfgfile)
    if config['pull'].get('refresh', None) is None:
        runonce = True

    logfile = config['core'].get('log', None)
    if config['core'].get('loglevel', 'info') == 'debug':
        loglevel = logging.DEBUG
    else:
        loglevel = logging.INFO

    if purge:
        # Override the pull.purge setting
        config['pull']['purge'] = 'yes'

    logger = grokmirror.init_logger('pull', logfile, loglevel, verbose)

    return pull_mirror(config, nomtime, forcepurge, runonce)


def command():
    opts = parse_args()

    retval = grok_pull(
        opts.config, opts.verbose, opts.nomtime, opts.purge, opts.forcepurge, opts.runonce)

    sys.exit(retval)


if __name__ == '__main__':
    command()