"""Run the importer as a service

Separate from importing and pushing individual packages, an importer service
maintains a set of repositories by repeatedly importing them when required.
"""

import collections
import datetime
import enum
import functools
import pprint
import sqlite3
import time


# The 'time' attribute is the publication date, as a timestamp, of the SPPHR
# corresponding to the URL stored in the 'link' attribute.
Timestamp = collections.namedtuple('Timestamp', ['time', 'link'])


class ImportRequest(
    collections.namedtuple('ImportRequest', ['package', 'level'])
):
    """Representation of an import request from the importer service

    :ivar str package: the name of the source package whose import is requested
    :ivar Level level: whether the request is for an incremental or full
        reimport
    """
    __slots__ = ()
    class Level(enum.Enum):
        # These numbers are hardcoded because they must match what we use in
        # our database schema.
        INCREMENTAL = 0
        REIMPORT = 1


def is_database_locked_exception(e):
    """Determine if a given exception is sqlite3's "database locked" exception

    :param Exception e: the exception to examine
    :rtype: bool
    :returns: True if the exception is sqlite3's "database is locked"
        exception, or False otherwise
    """
    # There doesn't seem to be any better way to detect the sqlite "locked"
    # exception than this.
    return (
        isinstance(e, sqlite3.OperationalError) and
            e.args[0] == 'database is locked'
    )


def _read_lock(method):
    """Method decorator to wrap database reads into a transaction

    To permit deadlock-free concurrency, all sqlite3 database operations need
    to be within explicit transactions. Transactions must not be left open as
    this will block other instances.

    Since the transactions implemented here cannot be reentrant, nothing
    inside the wrapped method can call a similarly decorated function. To
    prevent accidents, this is asserted using a "lock".

    The class that uses methods decorated by this function must initialize
    self._locked to False and provide a self.db attribute of type
    sqlite3.Connection that is the open sqlite3 connection constructed with
    sqlite3.connect(..., isolation_level=None).

    This decorator may retry the wrapped method multiple times in case of
    database locking timeouts. The wrapped method must accept this without
    error - for example by avoiding side effects or by being idempotent.

    :param callable method: the method to decorate.
    :raises AssertionError: (at wrapped method runtime) if a decorated
        method has already begun a transaction.
    :returns: the wrapped method
    """
    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        # Detect accidental reentrancy: a _read_lock or _write_lock
        # -decorated method must only appear once in the call stack.
        assert not self._locked
        self._locked = True
        try:
            # From https://sqlite.org/lang_transaction.html: "Internally, the
            # BEGIN DEFERRED statement merely sets a flag on the database
            # connection that turns off the automatic commit that would
            # normally occur when the last statement finishes."
            # This command should therefore never fail and so doesn't need
            # retrying on lock acquisition failure. We do explicitly call
            # BEGIN/ROLLBACK though in order to be robust against being wrapped
            # around a database modification by accident.
            self.db.execute('BEGIN DEFERRED')
            while True:
                try:
                    # We may get a "Database is locked" against the database
                    # during times of heavy system I/O when we called the
                    # wrapped function. In this case, retry the entire query.
                    return method(self, *args, **kwargs)
                except sqlite3.OperationalError as e:
                    if is_database_locked_exception(e):
                        continue
                    raise e
                finally:
                    self.db.execute('ROLLBACK')
        finally:
            self._locked = False
    return wrapper


def _write_lock(method):
    """Method decorator to wrap database writes into a transaction

    To permit deadlock-free concurrency, all sqlite3 database access need
    to be within explicit transactions and those transactions completed
    when not used. Further, transactions that write must begin with 'BEGIN
    IMMEDIATE'. This decorator makes this simple to do for methods in this
    class.

    Since the transactions implemented here cannot be reentrant, nothing
    inside the wrapped method can call a similarly decorated function. To
    prevent accidents, this is asserted using a "lock".

    The class that uses methods decorated by this function must initialize
    self._locked to False and provide a self.db attribute of type
    sqlite3.Connection that is the open sqlite3 connection constructed with
    sqlite3.connect(..., isolation_level=None).

    :param callable method: the method to decorate.
    :raises AssertionError: (at wrapped method runtime) if a decorated
        method has already begun a transaction.
    :returns: the wrapped method
    """
    @functools.wraps(method)
    def wrapper(self, *args, **kwargs):
        # Detect accidental reentrancy: a _read_lock or _write_lock
        # -decorated method must only appear once in the call stack.
        assert not self._locked
        self._locked = True
        try:
            while True:
                try:
                    self.db.execute('BEGIN IMMEDIATE')
                    break
                except sqlite3.OperationalError as e:
                    if is_database_locked_exception(e):
                        continue
                    raise e
            try:
                result = method(self, *args, **kwargs)
            except:
                self.db.execute('ROLLBACK')
                raise
            else:
                # We call 'COMMIT' explicitly instead of using
                # self.db.commit(). The latter is integrated with the
                # sqlite3's built-in transaction handling, which we've
                # disabled. How it behaves is therefore not precisely
                # defined, but what we do know is that it's safe for us to
                # take over all transaction-related commands explicitly
                # ourselves.
                self.db.execute('COMMIT')
                return result
        finally:
            self._locked = False
    return wrapper


class State(object):
    """Hold all persisted importer service state

    State is kept in an sqlite3 database.

    The public interface provides a data model as follows:

    A Timestamp (a namedtuple defined above) is maintained for each of
    ['debian', 'ubuntu'] which lists where we are up to in Launchpad's
    ever-growing list of source package publications for these distributions.
    read_timestamps() and write_timestamps() are used to read and write the
    timestamps.

    A list of outstanding "package import requests" is maintained. Requests are
    added using request_imports(). Each request is identified by a
    (source_package_name, timestamp) tuple. The timestamp nominally identifies
    the time the request was made. The request is treated as fulfilled when a
    package import has been completed and pushed to Launchpad and that import
    began after the request timestamp; this event is notified to this class
    using mark_import_outcomes(). If a request is made while an import is in
    progress, the request timestamp will be after the time the import began.
    Completion of the in progress import will not fulfil the request; an
    additional import run will be needed.

    Requests can be regular import requests, or reimport import requests.
    Reimport requests are mostly treated as separate requests, except that they
    can supersede regular import requests and fulfilling a reimport request
    also fulfils regular import requests (subject to timing of the fulfilment).

    Each package import request additionally has a status and a priority.

    The status is one of ['needed', 'done', 'error', 'superseded', 'running'].
    'done' means done successfully. If an import fails, the status will change
    to 'error'. 'done' and 'superseded' are not expected to be seen outside the
    implementation since 'superseded' and 'done' entries are removed
    automatically. These statuses exist to make the logic easier to follow
    during development, debugging and testing.

    A package import request is considered 'superseded' if a request with a
    higher timestamp exists for the same source package name. This includes
    running requests, which means that it is possible for a package import
    request to be running but have vanished from the database because it has
    been superseded by a newer one.

    The priority is an integer (low value means high priority) that can be used
    to order requests so that manual requests can be made that are prioritised
    over regular ones.
    """
    # Implementation details:

    # The request table is kept normalized before a commit with private method
    # State._normalize(), which detects and removes 'superseded' and 'done'
    # requests. If the database is committed without normalization then query
    # method results are undefined. Responsibility for keeping the database
    # committed normalized rests with this class. It is a bug in this class if
    # there exists any set of public method calls that results in changes
    # having been committed to the database without normalization.
    #
    # isolation_level=None, _read_lock() and _write_lock() are used to ensure
    # that concurrent database access (ie. multiple instances of this class;
    # one for the poller and one for the broker in the IPC case) works
    # deadlock-free. All database accesses must be wrapped in an explicit
    # transaction. See https://www.sqlite.org/lockingv3.html,
    # https://www.sqlite.org/isolation.html (particularly the "Isolation and
    # Concurrency" section) and https://sqlite.org/lang_transaction.html for
    # details. The transaction wrapping decorators are used on all public
    # methods. They are never used for internal methods, which assume that
    # transactions are already in progress as necessary. Tests may call
    # internal methods directly without transaction wrapping if they aren't
    # testing the transaction wrapping.

    LAST_SEEN_DISTROS = frozenset(['debian', 'ubuntu'])

    def __init__(self, db_path):
        # isolation_level=None, together with use of our _read_lock() and
        # _write_lock() decorators, is required to make concurrency work
        # deadlock-free
        self.db = sqlite3.connect(db_path, isolation_level=None)
        self._init_db(self.db)
        self._locked = False

    @classmethod
    def _init_db(cls, db):
        """Make sure the database has the schema defined

        :param sqlite3.Connection db: sqlite3 connection to use
        """
        # Instead of using the _write_lock() decorator, we wrap the transaction
        # manually in this method for this special case. self would be
        # unavailable anyway because this is a classmethod.
        db.execute('BEGIN IMMEDIATE')
        db.execute('''CREATE TABLE IF NOT EXISTS last_seen_publication (
            distro TEXT PRIMARY KEY,
            timestamp REAL NOT NULL,
            link TEXT
        )
        ''')
        db.execute('''CREATE TABLE IF NOT EXISTS request (
            srcpkg TEXT NOT NULL,
            timestamp REAL NOT NULL,
            status TEXT NOT NULL DEFAULT 'needed',
            priority INTEGER NOT NULL,
            reimport INTEGER NOT NULL DEFAULT 0,
            PRIMARY KEY (srcpkg, timestamp)
        )''')
        db.execute('''CREATE TABLE IF NOT EXISTS ipc_worker (
            identity TEXT NOT NULL,
            package TEXT NOT NULL,
            reimport INTEGER NOT NULL,
            reckoning_time INTEGER NOT NULL,
            PRIMARY KEY (identity)
        )''')

        # Handle schema migration. This can be dropped after all instances can
        # be expected to have migrated.
        cls._ensure_reimport_column(db)
        db.execute('COMMIT')

    @staticmethod
    def _ensure_reimport_column(db):
        """Add the reimport column to the request table if it doesn't exist

        The reimport column was added later so this method provides the schema
        migration. It can be dropped after all instances can be expected to
        have migrated.

        :param sqlite3.Connection db: the database to act upon
        :returns: None
        """
        cursor = db.cursor()
        cursor.execute('PRAGMA table_info(request)')

        # Determine which column number contains the column name
        col_name_index = {
            col[0]: i
            for i, col in enumerate(cursor.description)
        }['name']

        # Determine the set of column names present in the table
        columns = {row[col_name_index] for row in cursor.fetchall()}

        # Add the reimport column if it is not already present
        if 'reimport' not in columns:
            db.execute('''ALTER TABLE request ADD COLUMN
                reimport INTEGER NOT NULL DEFAULT 0
            ''')

    @_read_lock
    def _dump(self, table_name):
        """Return the contents of a table as a frozenset of tuples

        This is useful for testing.

        :returns: a set of tuples where the tuple is ordered the same as the
            request table definition.
        :rtype: depends on the table
        """
        cursor = self.db.cursor()
        # I wouldn't normally use * in case the column definition changes, but
        # in this case this method is defined to be identical to the table
        # definition and we will detect any mismatch by tests failing, so I
        # think in this case it makes sense.
        cursor.execute('SELECT * FROM %s' % table_name)
        return frozenset([tuple(row) for row in cursor.fetchall()])

    def _dump_request(self):
        """Return the contents of the request table as a frozenset of tuples

        This is useful for testing.

        :returns: a set of tuples where the tuple is ordered the same as the
            request table definition.
        :rtype: set(tuple(str, float, str, int, int))
        """
        return self._dump('request')

    def _dump_ipc_worker(self):
        """Return the contents of the request table as a frozenset of tuples

        This is useful for testing.

        :returns: a set of tuples where the tuple is ordered the same as the
            request table definition.
        :rtype: set(tuple(str, str, int, float))
        """
        return self._dump('ipc_worker')

    @staticmethod
    def _mark_superseded(db, assert_no_action=False):
        """Mark superseded request table rows as status='superseded'

        :param sqlite3.Connection db: sqlite3 connection to use
        :param bool assert_no_action: if True, assert that no changes were
            needed
        """
        cursor = db.cursor()

        # This SQL statement is a little obtuse but does exactly what we want:
        # mark superseded requests as superseded. For the purposes of this
        # statement, we're treating a request for a regular import for a
        # package and a request for a reimport for the same package as two
        # distinct requests.
        #
        # The meat is in the outer join. By joining to ourselves on
        # srcpkg=srcpkg, reimport=reimport and timestamp < timestamp, we end up
        # with every element of request on the left and something on the right
        # only when there exists a higher timestamp. By WHERE NOT NULL on that,
        # we eliminate rows where the request (on the left) has a higher
        # timestamp (ie. it isn't superseded). What's left (on the left) are
        # requests that have been superseded. This result is then fed to INSERT
        # OR REPLACE but with the status changed to 'superseded'. In practice
        # every attempted "INSERT" will conflict and be resolved with "REPLACE"
        # (thanks to the primary key). This has the effect of changing all the
        # rows that are superseded to having status set to 'superseded'.
        cursor.execute("""
        INSERT OR REPLACE INTO request
            (srcpkg, timestamp, status, priority, reimport)
            SELECT
                req_left.srcpkg, req_left.timestamp, 'superseded',
                    req_left.priority, req_left.reimport
                FROM request AS req_left LEFT OUTER JOIN request AS req_right
                ON req_left.srcpkg = req_right.srcpkg AND
                    req_left.reimport = req_right.reimport AND
                    req_left.timestamp < req_right.timestamp
                WHERE req_right.timestamp IS NOT NULL AND
                    req_left.status != 'running'
        """)
        assert not assert_no_action or not cursor.rowcount

        # If a regular import request for a package precedes (based on
        # timestamp) a reimport request for the same package, we can treat the
        # former request as superseded. This SQL statement marks such requests
        # as superseded using the same technique as the statement above.
        cursor.execute("""
        INSERT OR REPLACE INTO request
            (srcpkg, timestamp, status, priority, reimport)
            SELECT
                req_left.srcpkg, req_left.timestamp, 'superseded',
                    req_left.priority, req_left.reimport
                FROM request AS req_left LEFT OUTER JOIN request AS req_right
                ON req_left.srcpkg = req_right.srcpkg AND
                    req_left.timestamp < req_right.timestamp
                WHERE req_right.timestamp IS NOT NULL AND
                    req_left.status != 'running' AND
                    req_left.reimport = 0 AND
                    req_right.reimport = 1
        """)
        assert not assert_no_action or not cursor.rowcount

    @staticmethod
    def _delete_superseded(db, assert_no_action=False):
        """Delete request table rows marked status='superseded'

        :param sqlite3.Connection db: sqlite3 connection to use
        :param bool assert_no_action: if True, assert that no changes were
            needed
        """
        cursor = db.cursor()
        cursor.execute("DELETE FROM request WHERE status='superseded'")
        assert not assert_no_action or not cursor.rowcount

    @staticmethod
    def _delete_done(db, assert_no_action=False):
        """Delete request table rows marked status='done'

        status='done' applies only to successfully completed requests; requests
        that failed are marked status='error'.

        :param sqlite3.Connection db: sqlite3 connection to use
        :param bool assert_no_action: if True, assert that no changes were
            needed
        """
        cursor = db.cursor()
        cursor.execute("DELETE FROM request WHERE status='done'")
        assert not assert_no_action or not cursor.rowcount

    @classmethod
    def _normalize(cls, db, assert_no_action=False):
        """Normalize the database according to the defined normalization rules

        Normalization rules are defined in the docstring for this class.

        This method does not commit the changes. A database commit after all
        required changes are made is the caller's responsibility.

        :param sqlite3.Connection db: sqlite3 connection to use
        :param bool assert_no_action: if True, assert that no changes were
            needed
        """
        cls._mark_superseded(db=db, assert_no_action=assert_no_action)
        cls._delete_superseded(db=db, assert_no_action=assert_no_action)
        cls._delete_done(db=db, assert_no_action=assert_no_action)

    def _write_timestamps(self, timestamps):
        """Write timestamp values to data_directory

        This method is symmetrical to read_timestamps().

        :param dict(str, Timestamp) timestamps: the dictionary must contain
            'debian' and 'ubuntu' keys, each of which is a Timestamp
            namedtuple that corresponds to where we are up to in Launchpad
            against that distribution's publication history
        """
        assert set(timestamps.keys()) == self.LAST_SEEN_DISTROS
        for distro in self.LAST_SEEN_DISTROS:
            self.db.execute(
                '''INSERT OR REPLACE INTO last_seen_publication
                    (distro, timestamp, link)
                    VALUES (?, ?, ?)''',
                [distro, timestamps[distro].time, timestamps[distro].link],
            )

    @_read_lock
    def read_timestamps(self, num_days_ago=1, current_time=None):
        """Read saved timestamp values from data_directory

        This method is symmetrical to write_timestamps().

        :param: int num_days_ago: number of days ago to start iterating
            Launchpad publishes
        :param: float current_time: the "current" time to use for synthesis if
            no timestamps were previously stored. If None, then time.time() is
            used. This option is useful for testing.
        :returns: the timestamp dictionary previously stored with
            write_timestamps(), or, if timestamps were never previously stored,
            a timestamp dictionary synthesized based on current_time and
            num_days_ago and with links set to None
        :rtype: dict(str, Timestamp)
        """
        current_time = current_time or time.time()  # default: now

        cursor = self.db.cursor()
        cursor.execute(
            'SELECT distro, timestamp, link FROM last_seen_publication',
        )
        timestamps = {}
        for distro, _time, link in cursor:
            timestamps[distro] = Timestamp(time=_time, link=link)
        if not timestamps:
            _start = current_time - (num_days_ago * 24 * 60 * 60)
            return {
                distro: Timestamp(time=_start, link=None)
                for distro in self.LAST_SEEN_DISTROS
            }
        return timestamps

    @_write_lock
    def request_imports(
            self,
            pkgnames,
            timestamps=None,
            priority=20,
            request_time=None
        ):
        """Add package import requests to the needed import list

        :param iterable(str) pkgnames: source package names to import
        :param dict(str, Timestamp) timestamps: timestamps with which to update
            the last_seen_publication table
        :param int priority: the request priority (lower integer = higher
            priority)
        :param float request_time: the reckoning time of the request

        If timestamps is provided, then a call to write_timestamps is made and
        committed atomically at the same time as the request table update. This
        is useful when catching up new package import requests against new
        Launchpad publications. It is not expected to be used for ad-hoc
        package import requests.

        If request_time is provided then the time provided is noted as the time
        of the request instead of acquiring it using time.time(). This is
        useful for testing.
        """
        request_time = request_time or time.time()  # default: now
        self.db.executemany(
            '''INSERT OR REPLACE INTO request
                (srcpkg, priority, timestamp)
                VALUES (?, ?, ?)''',
            [(pkg, priority, request_time) for pkg in pkgnames]
        )
        self._normalize(self.db)
        if timestamps:
            self._write_timestamps(timestamps)

    @_read_lock
    def get_pending_imports(self):
        """Return a sequence of source package names that are pending import

        Source package names are returned in order of priority.

        Requests for reimports are treated as separate for requests for
        incremental imports, but if a package has both types of request marked
        as 'needed', only the reimport request will be returned. However, if a
        package has one type of request marked as 'running', this method will
        currently still return the other type of request as 'needed' if it
        exists.

        :rtype: sequence(ImportRequest)
        :returns: the package requests that are pending.
        """
        # The LEFT OUTER JOIN serves to "disqualify" any request that has a
        # matching request (one with the same srcpkg) that is already running.
        # Requests on the left are attempted to be matched with a disqualifying
        # request on the right. If a request on the left has no matching
        # disqualifying request, then the right side will be NULL. So filtering
        # the result where the right side is NULL gives us the qualifying
        # requests only.
        #
        # We then GROUP BY to collapse multiple requests for the same package
        # into one, such as if there is both a reimport and an incremental
        # import pending. max(req_left.reimport) causes us to favour reimport
        # (reimport=1) over incremental (reimport=0).
        cursor = self.db.cursor()
        cursor.execute("""SELECT req_left.srcpkg, max(req_left.reimport)
            FROM request AS req_left LEFT OUTER JOIN request AS req_right
            ON req_left.srcpkg = req_right.srcpkg AND
                req_right.status == 'running'
            WHERE req_right.srcpkg IS NULL AND req_left.status = 'needed'
            GROUP BY req_left.srcpkg
            ORDER BY req_left.priority
        """)
        return [
            ImportRequest(package=package, level=ImportRequest.Level(reimport))
            for package, reimport in cursor
        ]

    @_write_lock
    def mark_running(self, requests):
        self._mark_running(requests)

    def _mark_running(self, requests):
        """Mark packages as being imported

        Change the status to 'running' for each of the listed import requests.
        This list must be the set or a subset of packages listed by a previous
        call to get_pending_imports().

        :param sequence(ImportRequest) requests: requests to mark as running
        :returns: None
        """
        self.db.executemany(
            """
                UPDATE request SET status='running'
                    WHERE srcpkg=? and reimport=?""",
            ((request.package, request.level.value) for request in requests),
        )
        self._normalize(self.db)

    @_write_lock
    def unmark_running(self):
        """Mark packages marked as running as failed

        Change the status to 'error' for every package that is currently listed
        as running. This can be used, for example, on restart, where we know
        that there are no packages running so they must all have failed.
        """

        self.db.execute(
            "UPDATE request SET status='error' WHERE status='running'"
        )
        self._normalize(self.db)

    @_write_lock
    def mark_import_outcomes(
            self,
            reckoning_time,
            completed_requests,
            failed_requests,
        ):
        """Mark some package import requests as complete

        :param float reckoning_time: a time prior to the time the imports
            began. Only requests whose timestamps are before this time are
            considered to have been completed by the report made by this method
            call.
        :param sequence(ImportRequest) completed_requests: requests whose
            corresponding imports have completed successfully.
        :param sequence(ImportRequest) failed_pkgs: requests whose
            corresponding imports have failed.
        """
        # A reimport also fulfils an incremental import request. However an
        # incremental import does not fulfil a reimport request. Since reimport
        # is an integer (either 0 or 1) in sqlite3, we can use the '<='
        # operator to mark as done any request that is fulfilled but avoid
        # marking a reimport request as done if we only completed an
        # incremental import.
        self.db.executemany(
            """UPDATE request
                SET status='done'
                WHERE srcpkg=? AND reimport <= ? AND timestamp < ?""",
            [
                (request.package, request.level.value, reckoning_time)
                for request in completed_requests
            ],
        )

        # To mark failed requests as errored, we don't need do any "one type
        # fulfils the other type" thinking like above. We just match exactly
        # against the request and mark that one as errored.
        self.db.executemany(
            """UPDATE request
                SET status='error'
                WHERE srcpkg=? AND reimport = ? AND timestamp < ?""",
            [
                (request.package, request.level.value, reckoning_time)
                for request in failed_requests
            ],
        )
        self._normalize(self.db)

    @_write_lock
    def assign_ipc_worker(self, identity, request, reckoning_time=None):
        """Mark a request as in progress by an IPC worker

        This will mark the request as "running" in the request table and also
        add an entry to the ipc_worker table in a single transaction.

        :param str identity: the identity of the worker (an arbitrary string)
        :param ImportRequest request: the import request the worker will work
            on
        :param float reckoning_time: the latest point in time that is
            definitely before the worker begins work on this request. If None,
            then the current time is used.
        :returns: None
        """
        if reckoning_time is None:
            reckoning_time = time.time()
        self.db.execute(
            '''INSERT INTO ipc_worker
                (identity, package, reimport, reckoning_time)
                VALUES (?, ?, ?, ?)''',
            (identity, request.package, request.level.value, reckoning_time),
        )
        self._mark_running([request])

    @_read_lock
    def lookup_ipc_worker(self, identity):
        """Find details of a request that is in progress by an IPC worker

        The request must exist in the ipc_worker table.

        :param str identity: the identity of the worker (an arbitrary string)
        :rtype: tuple(str, ImportRequest.Level, float)
        :returns: a tuple of the package name, the import request level, and
            the reckoning time, as previously stored in the ipc_worker table.
        """
        cursor = self.db.cursor()
        cursor.execute('''
                SELECT package, reimport, reckoning_time
                FROM ipc_worker
                WHERE identity=?''',
            (identity,),
        )
        row = cursor.fetchone()
        assert row
        package, reimport, reckoning_time = row
        return package, ImportRequest.Level(reimport), reckoning_time

    @_write_lock
    def clear_ipc_worker(self, identity):
        """Clear our record of work being done by a particular worker

        Unmark a worker from working on a particular request. If the request is
        complete, it is assumed that the request itself has been marked as
        'done' already.

        If we knew about the given worker and it is recorded as still working
        on a request, mark the worker as no longer working on that request and
        that request as having errored. This provides a "self-healing" property
        in case the worker and broker end up out-of-sync (eg. if one or the
        other is restarted).

        If the given worker is not recorded as currently doing any work, then
        this method does nothing and succeeds.

        :param str identity: the identity of the worker (an arbitrary string)
        :returns: None
        """
        cursor = self.db.cursor()
        cursor.execute(
            'SELECT package, reimport FROM ipc_worker WHERE identity=?',
            (identity,),
        )
        row = cursor.fetchone()
        if row:
            package, reimport = row
            cursor.execute("""
                    UPDATE request
                    SET status='error'
                    WHERE srcpkg=? AND status='running' AND reimport=?""",
                (package, reimport),
            )
            cursor.execute(
                'DELETE FROM ipc_worker WHERE identity=?',
                (identity,),
            )


def read_package_list(path):
    """Read a list of package names from a text file

    The file must contain one package name per line. Whitespace is stripped.
    Blank lines are permitted. '#' is the comment character: lines that begin
    with '#' (ignoring whitespace) are ignored. Comments are not currently
    permitted on the same line as a package name.

    :param str path: the path to the text file
    :rtype: list(str)
    :returns: the list of packages in the given file
    """
    try:
        fobj = open(path, 'r')
    except FileNotFoundError:
        return []
    with fobj:
        return [
            line.strip() for line in fobj
            if not line.startswith('#')
        ]


def has_import_repository(lp, team, package):
    """Determine if an import repository exists for a given package

    :param launchpadlib.launchpad.Launchpad lp: Launchpad object
    :param str team: name of Launchpad team that contains imported repositories
    :param str package: source package name
    :rtype: bool
    :returns: True if the import repository exists, or False otherwise
    """
    return bool(
        lp.git_repositories.getByPath(
            path='~%s/ubuntu/+source/%s' % (team, package),
        ),
    )


def request_new_imports(
    lp,
    num_days_ago,
    service_state,
    filter_func=None,
    request_time=None,
):
    """Import all new publications since a prior execution

    The purpose of this function is the side-effect of moving along
    service_state's last_seen_publication timestamps, sweeping any publications
    found along the way into import requests as required.

    :param launchpadlib.launchpad.Launchpad lp: the Launchpad connection to use
        to query Launchpad.
    :param int num_days_ago: how many days ago to start iterating Launchpad
        publications at if we've never looked at Launchpad publications before
    :param importer_service.State service_state: the persistent data on
        the status of the import
    :param filter_func: a function that when called with a sole parameter
        string providing a package name, returns a bool that specifies whether
        or not that package should be imported.
    :param float request_time: the reckoning time of the generated requests. If
        this evaluates to False, use the current time. Useful for testing.
    :returns: None
    """
    timestamps = service_state.read_timestamps(num_days_ago)

    # filtered_pkgnames is the list of source package names across all
    # distributions we will want to process, based upon
    # scriptutils.should_import_srcpkg
    filtered_pkgnames = set()

    for dist_name in ['debian', 'ubuntu']:
        timestamp = timestamps[dist_name]

        # dist_newest_spphr is the most recent publication record with a
        # valid published date in the distribution
        dist_newest_spphr = None

        # dist_filtered_pkgnames is the set of source package names in
        # the dist_name distribution that qualify for import according to our
        # allowlists, denylists and phasing requirements
        dist_filtered_pkgnames = set()

        print(
            "Examining publications in %s since %s" % (
                dist_name,
                datetime.datetime.fromtimestamp(
                    timestamp.time,
                ).strftime("%Y-%m-%d %H:%M:%S"),
            )
        )

        # spph is the raw publication history for a distribution from
        # Launchpad, sorted by publication date in reverse chronological order
        dist = lp.distributions[dist_name]
        spph = dist.main_archive.getPublishedSources(order_by_date=True)
        if not spph:
            print("No publishing data found in %s" % dist_name)
            continue

        for spphr in spph:
            # this is the matching upload to the last publish seen
            if str(spphr) == timestamp.link:
                break

            # only check if we should stop iterating due to the
            # timestamps if there is a timestamp to compare to, which
            # means the source package is actually published.
            if spphr.date_published:
                if not dist_newest_spphr:
                    dist_newest_spphr = spphr
                # stop iterating (backwards chronologically) when we see
                # a publish timestamp before the last run
                if spphr.date_published.timestamp() < timestamp.time:
                    break

                package = spphr.source_package_name
                if filter_func and filter_func(package):
                    dist_filtered_pkgnames.add(package)

        if not dist_filtered_pkgnames:
            print(
                "No new relevant publications found in %s relative to %s" % (
                    dist_name,
                    datetime.datetime.fromtimestamp(
                        timestamp.time
                    ).strftime("%Y-%m-%d %H:%M:%S"),
                )
            )

        filtered_pkgnames = filtered_pkgnames | dist_filtered_pkgnames
        if dist_newest_spphr:
            # Update timestamp
            timestamps[dist_name] = timestamp = Timestamp(
                time=dist_newest_spphr.date_published.timestamp(),
                link=str(dist_newest_spphr),
            )

    service_state.request_imports(
        pkgnames=filtered_pkgnames,
        timestamps=timestamps,
        request_time=request_time,
    )
    print("Requested imports for:")
    pprint.pprint(filtered_pkgnames)