# -*- coding: utf-8 -*-

"""

 (c) 2017 - Copyright Red Hat Inc

 Authors:

   Patrick Uiterwijk <puiterwijk@redhat.com></puiterwijk@redhat.com>

"""

from __future__ import unicode_literals

import collections

import datetime

import gc

import hashlib

import os

import os.path

import shutil

import subprocess

import time

from functools import wraps

import arrow

import pygit2

import six

from celery import Celery

from celery.result import AsyncResult

from celery.signals import after_setup_task_logger

from celery.utils.log import get_task_logger

from sqlalchemy.exc import SQLAlchemyError

import pagure.lib.git

import pagure.lib.git_auth

import pagure.lib.link

import pagure.lib.query

import pagure.lib.repo

import pagure.utils

from pagure.config import config as pagure_config

from pagure.utils import get_parent_repo_path

# logging.config.dictConfig(pagure_config.get('LOGGING') or {'version': 1})

_log = get_task_logger(__name__)

if os.environ.get("PAGURE_BROKER_URL"):

    broker_url = os.environ["PAGURE_BROKER_URL"]

elif pagure_config.get("BROKER_URL"):

    broker_url = pagure_config["BROKER_URL"]

else:

    broker_url = "redis://%s" % pagure_config["REDIS_HOST"]

conn = Celery("tasks", broker=broker_url, backend=broker_url)

conn.conf.update(pagure_config["CELERY_CONFIG"])

@after_setup_task_logger.connect

def augment_celery_log(**kwargs):

    pagure.utils.set_up_logging(force=True)

def pagure_task(function):

    """ Simple decorator that is responsible for:

    * Adjusting the status of the task when it starts

    * Creating and cleaning up a SQLAlchemy session

    """

    @wraps(function)

    def decorated_function(self, *args, **kwargs):

        """ Decorated function, actually does the work. """

        if self is not None:

            try:

                self.update_state(state="RUNNING")

            except TypeError:

                pass

        session = pagure.lib.query.create_session(pagure_config["DB_URL"])

        try:

            return function(self, session, *args, **kwargs)

        except:  # noqa: E722

            # if the task has raised for any reason, we need to rollback the

            # session first to not leave open uncomitted transaction hanging

            session.rollback()

            raise

        finally:

            session.remove()

            gc_clean()

    return decorated_function

def get_result(uuid):

    """ Returns the AsyncResult object for a given task.

    :arg uuid: the unique identifier of the task to retrieve.

    :type uuid: str

    :return: celery.result.AsyncResult

    """

    return AsyncResult(uuid, conn.backend)

def ret(endpoint, **kwargs):

    toret = {"endpoint": endpoint}

    toret.update(kwargs)

    return toret

def gc_clean():

    """ Force a run of the garbage collector. """

    # https://pagure.io/pagure/issue/2302

    gc.collect()

@conn.task(queue=pagure_config.get("GITOLITE_CELERY_QUEUE", None), bind=True)

@pagure_task

def generate_gitolite_acls(

    self, session, namespace=None, name=None, user=None, group=None

):

    """ Generate the gitolite configuration file either entirely or for a

    specific project.

    :arg session: SQLAlchemy session object

    :type session: sqlalchemy.orm.session.Session

    :kwarg namespace: the namespace of the project

    :type namespace: None or str

    :kwarg name: the name of the project

    :type name: None or str

    :kwarg user: the user of the project, only set if the project is a fork

    :type user: None or str

    :kwarg group: the group to refresh the members of

    :type group: None or str

    """

    project = None

    if name and name != -1:

        project = pagure.lib.query._get_project(

            session, namespace=namespace, name=name, user=user

        )

    elif name == -1:

        project = name

    helper = pagure.lib.git_auth.get_git_auth_helper()

    _log.debug("Got helper: %s", helper)

    group_obj = None

    if group:

        group_obj = pagure.lib.query.search_groups(session, group_name=group)

    _log.debug(

        "Calling helper: %s with arg: project=%s, group=%s",

        helper,

        project,

        group_obj,

    )

    helper.generate_acls(project=project, group=group_obj)

    pagure.lib.query.update_read_only_mode(session, project, read_only=False)

    try:

        session.commit()

        _log.debug("Project %s is no longer in Read Only Mode", project)

    except SQLAlchemyError:

        session.rollback()

        _log.exception("Failed to unmark read_only for: %s project", project)

@conn.task(queue=pagure_config.get("GITOLITE_CELERY_QUEUE", None), bind=True)

@pagure_task

def gitolite_post_compile_only(self, session):

    """ Do gitolite post-processing only. Most importantly, this processes SSH

    keys used by gitolite. This is an optimization task that's supposed to be

    used if you only need to run `gitolite trigger POST_COMPILE` without

    touching any other gitolite configuration

    """

    helper = pagure.lib.git_auth.get_git_auth_helper()

    _log.debug("Got helper: %s", helper)

    if hasattr(helper, "post_compile_only"):

        helper.post_compile_only()

    else:

        helper.generate_acls(project=None)

@conn.task(queue=pagure_config.get("GITOLITE_CELERY_QUEUE", None), bind=True)

@pagure_task

def delete_project(

    self, session, namespace=None, name=None, user=None, action_user=None

):

    """ Delete a project in pagure.

    This is achieved in three steps:

    - Remove the project from gitolite.conf

    - Remove the git repositories on disk

    - Remove the project from the DB

    :arg session: SQLAlchemy session object

    :type session: sqlalchemy.orm.session.Session

    :kwarg namespace: the namespace of the project

    :type namespace: None or str

    :kwarg name: the name of the project

    :type name: None or str

    :kwarg user: the user of the project, only set if the project is a fork

    :type user: None or str

    :kwarg action_user: the user deleting the project

    :type action_user: None or str

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    if not project:

        raise RuntimeError(

            "Project: %s/%s from user: %s not found in the DB"

            % (namespace, name, user)

        )

    # Remove the project from gitolite.conf

    helper = pagure.lib.git_auth.get_git_auth_helper()

    _log.debug("Got helper: %s", helper)

    _log.debug(

        "Calling helper: %s with arg: project=%s", helper, project.fullname

    )

    helper.remove_acls(session=session, project=project)

    # Remove the git repositories on disk

    pagure.lib.git.delete_project_repos(project)

    # Remove the project from the DB

    username = project.user.user

    try:

        project_json = project.to_json(public=True)

        session.delete(project)

        session.commit()

        pagure.lib.notify.log(

            project,

            topic="project.deleted",

            msg=dict(project=project_json, agent=action_user),

        )

    except SQLAlchemyError:

        session.rollback()

        _log.exception(

            "Failed to delete project: %s from the DB", project.fullname

        )

    return ret("ui_ns.view_user", username=username)

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def create_project(

    self, session, username, namespace, name, add_readme, ignore_existing_repo

):

    """ Create a project.

    :arg session: SQLAlchemy session object

    :type session: sqlalchemy.orm.session.Session

    :kwarg username: the user creating the project

    :type user: str

    :kwarg namespace: the namespace of the project

    :type namespace: str

    :kwarg name: the name of the project

    :type name: str

    :kwarg add_readme: a boolean specifying if the project should be

        created with a README file or not

    :type add_readme: bool

    :kwarg ignore_existing_repo: a boolean specifying whether the creation

        of the project should fail if the repo exists on disk or not

    :type ignore_existing_repo: bool

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name

    )

    userobj = pagure.lib.query.search_user(session, username=username)

    # Add the readme file if it was asked

    templ = None

    if project.is_fork:

        templ = pagure_config.get("FORK_TEMPLATE_PATH")

    else:

        templ = pagure_config.get("PROJECT_TEMPLATE_PATH")

    if templ:

        if not os.path.exists(templ):

            _log.warning(

                "Invalid git template configured: %s, not found on disk", templ

            )

            templ = None

        else:

            _log.debug("  Using template at: %s", templ)

    # There is a risk for a race-condition here between when the repo is

    # created and when the README gets added. However, this risk is small

    # enough that we will keep this as is for now (esp since it fixes the

    # situation where deleting the project raised an error if it was in the

    # middle of the lock)

    try:

        with project.lock("WORKER"):

            pagure.lib.git.create_project_repos(

                project,

                project.repospanner_region,

                templ,

                ignore_existing_repo,

            )

    except pagure.exceptions.RepoExistsException:

        session.delete(project)

        session.commit()

        raise

    if add_readme:

        with project.lock("WORKER"):

            with pagure.lib.git.TemporaryClone(

                project, "main", "add_readme"

            ) as tempclone:

                temp_gitrepo = tempclone.repo

                # Add README file

                author = userobj.fullname or userobj.user

                author_email = userobj.default_email

                if six.PY2:

                    author = author.encode("utf-8")

                    author_email = author_email.encode("utf-8")

                author = pygit2.Signature(author, author_email)

                content = "# %s\n\n%s" % (name, project.description)

                readme_file = os.path.join(temp_gitrepo.workdir, "README.md")

                with open(readme_file, "wb") as stream:

                    stream.write(content.encode("utf-8"))

                temp_gitrepo.index.add_all()

                temp_gitrepo.index.write()

                tree = temp_gitrepo.index.write_tree()

                temp_gitrepo.create_commit(

                    "HEAD", author, author, "Added the README", tree, []

                )

                master_ref = temp_gitrepo.lookup_reference("HEAD").resolve()

                tempclone.push("pagure", master_ref.name, internal="yes")

            # Install the default hook

            plugin = pagure.lib.plugins.get_plugin("default")

            dbobj = plugin.db_object()

            dbobj.active = True

            dbobj.project_id = project.id

            session.add(dbobj)

            session.flush()

            plugin.set_up(project)

            plugin.install(project, dbobj)

            session.commit()

    task = generate_gitolite_acls.delay(

        namespace=project.namespace,

        name=project.name,

        user=project.user.user if project.is_fork else None,

    )

    _log.info("Refreshing gitolite config queued in task: %s", task.id)

    return ret("ui_ns.view_repo", repo=name, namespace=namespace)

@conn.task(queue=pagure_config.get("SLOW_CELERY_QUEUE", None), bind=True)

@pagure_task

def update_git(

    self, session, name, namespace, user, ticketuid=None, requestuid=None

):

    """ Update the JSON representation of either a ticket or a pull-request

    depending on the argument specified.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    project_lock = "WORKER"

    if ticketuid is not None:

        project_lock = "WORKER_TICKET"

    elif requestuid is not None:

        project_lock = "WORKER_REQUEST"

    with project.lock(project_lock):

        if ticketuid is not None:

            obj = pagure.lib.query.get_issue_by_uid(session, ticketuid)

        elif requestuid is not None:

            obj = pagure.lib.query.get_request_by_uid(session, requestuid)

        else:

            raise NotImplementedError("No ticket ID or request ID provided")

        if obj is None:

            raise Exception("Unable to find object")

        result = pagure.lib.git._update_git(obj, project)

    return result

@conn.task(queue=pagure_config.get("SLOW_CELERY_QUEUE", None), bind=True)

@pagure_task

def clean_git(self, session, name, namespace, user, obj_repotype, obj_uid):

    """ Remove the JSON representation of a ticket on the git repository

    for tickets.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER_TICKET"):

        result = pagure.lib.git._clean_git(project, obj_repotype, obj_uid)

    return result

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def update_file_in_git(

    self,

    session,

    name,

    namespace,

    user,

    branch,

    branchto,

    filename,

    content,

    message,

    username,

    email,

):

    """ Update a file in the specified git repo.

    """

    userobj = pagure.lib.query.search_user(session, username=username)

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        pagure.lib.git._update_file_in_git(

            project,

            branch,

            branchto,

            filename,

            content,

            message,

            userobj,

            email,

        )

    return ret(

        "ui_ns.view_commits",

        repo=project.name,

        username=user,

        namespace=namespace,

        branchname=branchto,

    )

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def delete_branch(self, session, name, namespace, user, branchname):

    """ Delete a branch from a git repo.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        repo_obj = pygit2.Repository(pagure.utils.get_repo_path(project))

        try:

            branch = repo_obj.lookup_branch(branchname)

            branch.delete()

        except pygit2.GitError as err:

            _log.exception(err)

    return ret(

        "ui_ns.view_branches", repo=name, namespace=namespace, username=user

    )

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def fork(

    self,

    session,

    name,

    namespace,

    user_owner,

    user_forker,

    editbranch,

    editfile,

):

    """ Forks the specified project for the specified user.

    :arg namespace: the namespace of the project

    :type namespace: str

    :arg name: the name of the project

    :type name: str

    :arg user_owner: the user of which the project is forked, only set

        if the project is already a fork

    :type user_owner: str

    :arg user_forker: the user forking the project

    :type user_forker: str

    :kwarg editbranch: the name of the branch in which the user asked to

        edit a file

    :type editbranch: str

    :kwarg editfile: the file the user asked to edit

    :type editfile: str

    """

    repo_from = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user_owner

    )

    repo_to = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user_forker

    )

    with repo_to.lock("WORKER"):

        pagure.lib.git.create_project_repos(

            repo_to, repo_to.repospanner_region, None, False

        )

        with pagure.lib.git.TemporaryClone(

            repo_to, "main", "fork"

        ) as tempclone:

            fork_repo = tempclone.repo

            fork_repo.remotes.create("forkedfrom", repo_from.repopath("main"))

            fork_repo.remotes["forkedfrom"].fetch()

            for branchname in fork_repo.branches.remote:

                if not branchname.startswith("forkedfrom/"):

                    continue

                localname = branchname.replace("forkedfrom/", "")

                if localname == "HEAD":

                    # HEAD will be created automatically as a symref

                    continue

                tempclone.push(

                    "pagure",

                    "remotes/%s" % branchname,

                    "refs/heads/%s" % localname,

                    internal="yes",

                )

        if not repo_to.is_on_repospanner and not repo_to.private:

            # Create the git-daemon-export-ok file on the clone

            http_clone_file = os.path.join(

                repo_to.repopath("main"), "git-daemon-export-ok"

            )

            if not os.path.exists(http_clone_file):

                with open(http_clone_file, "w"):

                    pass

        pagure.lib.notify.log(

            repo_to,

            topic="project.forked",

            msg=dict(project=repo_to.to_json(public=True), agent=user_forker),

        )

    _log.info("Project created, refreshing auth async")

    task = generate_gitolite_acls.delay(

        namespace=repo_to.namespace,

        name=repo_to.name,

        user=repo_to.user.user if repo_to.is_fork else None,

    )

    _log.info("Refreshing gitolite config queued in task: %s", task.id)

    if editfile is None:

        return ret(

            "ui_ns.view_repo",

            repo=name,

            namespace=namespace,

            username=user_forker,

        )

    else:

        return ret(

            "ui_ns.edit_file",

            repo=name,

            namespace=namespace,

            username=user_forker,

            branchname=editbranch,

            filename=editfile,

        )

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def pull_remote_repo(self, session, remote_git, branch_from):

    """ Clone a remote git repository locally for remote PRs.

    """

    clonepath = pagure.utils.get_remote_repo_path(

        remote_git, branch_from, ignore_non_exist=True

    )

    repo = pygit2.clone_repository(

        remote_git, clonepath, checkout_branch=branch_from

    )

    del repo

    return clonepath

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def refresh_remote_pr(self, session, name, namespace, user, requestid):

    """ Refresh the local clone of a git repository used in a remote

    pull-request.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    request = pagure.lib.query.search_pull_requests(

        session, project_id=project.id, requestid=requestid

    )

    _log.debug(

        "refreshing remote pull-request: %s/#%s",

        request.project.fullname,

        request.id,

    )

    clonepath = pagure.utils.get_remote_repo_path(

        request.remote_git, request.branch_from

    )

    repo = pagure.lib.repo.PagureRepo(clonepath)

    repo.pull(branch=request.branch_from, force=True)

    refresh_pr_cache.delay(name, namespace, user)

    del repo

    return ret(

        "ui_ns.request_pull",

        username=user,

        namespace=namespace,

        repo=name,

        requestid=requestid,

    )

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def move_to_repospanner(self, session, name, namespace, user, region):

    """ Move a repository to a repoSpanner region.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    regioninfo = pagure_config.get("REPOSPANNER_REGIONS", {}).get(region)

    if not regioninfo:

        raise Exception("Missing region config")

    with project.lock("WORKER"):

        # Perform some pre-flight checks

        if project.is_on_repospanner:

            raise Exception("Project is already on repoSpanner")

        #  Make sure that no non-runner hooks are enabled for this project

        compatible_targets = [pagure.lib.query.HOOK_DNE_TARGET]

        incompatible_hooks = []

        for repotype in pagure.lib.query.REPOTYPES:

            path = project.repopath(repotype)

            if path is None:

                continue

            hookpath = os.path.join(path, "hooks")

            for hook in os.listdir(hookpath):

                if not hook.startswith(

                    ("pre-receive.", "update.", "post-receive.")

                ):

                    continue

                if hook.endswith(".sample"):

                    # Ignore the samples that Git inserts

                    continue

                hookfile = os.path.join(hookpath, hook)

                if os.path.realpath(hookfile) not in compatible_targets:

                    incompatible_hooks.append((repotype, hook))

        if incompatible_hooks:

            raise Exception(

                "Repository contains repoSpanner-incompatible "

                "hooks: %s"

                % ", ".join(["%s" % (hook,) for hook in incompatible_hooks])

            )

        # Create the repositories

        pagure.lib.git.create_project_repos(project, region, None, False)

        for repotype in pagure.lib.query.REPOTYPES:

            repopath = project.repopath(repotype)

            if repopath is None:

                continue

            repourl, _ = project.repospanner_repo_info(repotype, region)

            repo_obj = pagure.lib.repo.PagureRepo(repopath)

            repo_obj.create_remote("repospanner_push", repourl)

            command = [

                "git",

                "-c",

                "http.sslcainfo=%s" % regioninfo["ca"],

                "-c",

                "http.sslcert=%s" % regioninfo["push_cert"]["cert"],

                "-c",

                "http.sslkey=%s" % regioninfo["push_cert"]["key"],

                "push",

                "--mirror",

                "repospanner_push",

            ]

            _log.debug("Running push command: %s", command)

            out = subprocess.check_output(

                command, cwd=repopath, stderr=subprocess.STDOUT

            )

            _log.debug("Out: %s" % out)

        for repotype in pagure.lib.query.REPOTYPES:

            repopath = project.repopath(repotype)

            if repopath is None:

                continue

            repo_obj = pagure.lib.repo.PagureRepo(repopath)

            # At this moment, this subrepo has been migrated

            # Move the "refs" folder to "refsold", so that we don't actually

            # delete any data, but it's no longer a valid git repo.

            # On next use, a pseudo repository will be created.

            refsdir = os.path.join(repopath, "refs")

            shutil.move(refsdir, refsdir + "old")

            with open(

                os.path.join(repopath, "repospanner_status"), "w"

            ) as info:

                info.write(

                    "This repository has migrated to repoSpanner region %s"

                    % region

                )

        project.repospanner_region = region

        session.add(project)

        session.commit()

    return ret(

        "ui_ns.view_repo", username=user, namespace=namespace, repo=name

    )

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def refresh_pr_cache(self, session, name, namespace, user):

    """ Refresh the merge status cached of pull-requests.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    pagure.lib.query.reset_status_pull_request(session, project)

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def merge_pull_request(

    self,

    session,

    name,

    namespace,

    user,

    requestid,

    user_merger,

    delete_branch_after=False,

):

    """ Merge pull-request.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        request = pagure.lib.query.search_pull_requests(

            session, project_id=project.id, requestid=requestid

        )

        _log.debug(

            "Merging pull-request: %s/#%s",

            request.project.fullname,

            request.id,

        )

        pagure.lib.git.merge_pull_request(session, request, user_merger)

    if delete_branch_after:

        _log.debug(

            "Will delete source branch of pull-request: %s/#%s",

            request.project.fullname,

            request.id,

        )

        owner = (

            request.project_from.user.username

            if request.project_from.parent

            else None

        )

        delete_branch.delay(

            request.project_from.name,

            request.project_from.namespace,

            owner,

            request.branch_from,

        )

    refresh_pr_cache.delay(name, namespace, user)

    return ret(

        "ui_ns.view_repo", repo=name, username=user, namespace=namespace

    )

@conn.task(queue=pagure_config.get("FAST_CELERY_QUEUE", None), bind=True)

@pagure_task

def add_file_to_git(

    self, session, name, namespace, user, user_attacher, issueuid, filename

):

    """ Add a file to the specified git repo.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        issue = pagure.lib.query.get_issue_by_uid(session, issueuid)

        user_attacher = pagure.lib.query.search_user(

            session, username=user_attacher

        )

        from_folder = pagure_config["ATTACHMENTS_FOLDER"]

        _log.info(

            "Adding file %s from %s to %s",

            filename,

            from_folder,

            project.fullname,

        )

        pagure.lib.git._add_file_to_git(

            project, issue, from_folder, user_attacher, filename

        )

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def project_dowait(self, session, name, namespace, user):

    """ This is a task used to test the locking systems.

    It should never be allowed to be called in production instances, since that

    would allow an attacker to basically DOS a project by calling this

    repeatedly. """

    assert pagure_config.get("ALLOW_PROJECT_DOWAIT", False)

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        time.sleep(10)

    return ret(

        "ui_ns.view_repo", repo=name, username=user, namespace=namespace

    )

@conn.task(queue=pagure_config.get("MEDIUM_CELERY_QUEUE", None), bind=True)

@pagure_task

def sync_pull_ref(self, session, name, namespace, user, requestid):

    """ Synchronize a pull/ reference from the content in the forked repo,

    allowing local checkout of the pull-request.

    """

    project = pagure.lib.query._get_project(

        session, namespace=namespace, name=name, user=user

    )

    with project.lock("WORKER"):

        request = pagure.lib.query.search_pull_requests(

            session, project_id=project.id, requestid=requestid

        )

        _log.debug(

            "Update pull refs of: %s#%s", request.project.fullname, request.id

        )

        if request.remote:

            # Get the fork

            repopath = pagure.utils.get_remote_repo_path(

                request.remote_git, request.branch_from

            )

        else: