docs/source/plugins.rst

@@ -281,6 +281,61 @@ Admins should consider the balance between the ``batch_size`` and
 plugin could attempt to send during a single call.
 
 
+Message queue
+=============
+
+The ``message`` plugin for the hub enables a dedicated client to implement
+message bus connections.
+
+It works by recording messages about notable hub activities
+in the database.
+A dedicated client can use the ``message.poll()`` and ``message.ack()`` methods
+to fetch and acknowledge messages from the queue. These calls require the new
+``message`` permission.
+Most callback events on the hub are translated into messages.
+
+In order to enable this plugin, you must add ``message`` to the ``Plugins`` setting in ``/etc/koji-hub/hub.conf``
+
+.. warning:: Do not enable this plugin unless you have a dedicated client that
+   will consume the queue.
+
+This plugin can be configured in the kojihub configuration file(s).
+To do so, add a ``[plugin.message]`` section.
+The following configuration options are available:
+
+* ``string_limit`` -- the maximum allowed length for string values that
+  appear in messages. If a string field is longer, then it will be omitted.
+  The default value is ``0`` which means no limit.
+* ``item_limit`` -- the maximum allowed length for list and dictionary values that
+  appear in messages. If a field is longer, then it will be omitted.
+  Note that lists are never truncated. The entire list is dropped if it is too long.
+  The default value is ``0`` which means no limit.
+* ``batch_size`` -- the maximum number of queued messages to report per call.
+  The default is 100.
+* ``max_age`` -- the age (in hours) at which old messages in the queue are
+  discarded. The default is 24.
+* ``max_sent_age`` -- the age (in hours) at which sent messages in the queue
+  are discarded. By default, sent messages have the same lifespan as unsent.
+* ``auto_clean`` -- (boolean) whether to automatically clean the queue
+  periodically. Default: true.
+* ``auto_clean_interval`` -- the number of seconds to wait between automatic
+  cleanings. Default: 300.
+
+API calls
+---------
+
+The message plugin provides three api calls for managing the message queue.
+All of these calls require the ``message`` permission.
+
+* ``message.poll(limit=None)`` -- returns unsent messages from the queue. These
+  messages remain in the unsent state until acknowledged.
+* ``message.ack(msg_ids)`` -- acknowledge messages. The messages are marked
+  sent and will no longer be reported by the poll call. If ``auto_clean`` is
+  enabled, this will also periodically trigger cleanup.
+* ``message.clean()`` -- explicitly trigger queue cleanup. This should only be
+  needed if ``auto_clean`` is disabled.
+
+
 Image builds using Kiwi
 =======================
 

kojihub/kojixmlrpc.py

@@ -529,24 +529,19 @@ config_map = [
 def load_config(environ):
     """Load configuration options
 
-    Options are read from a config file. The config file location is
-    controlled by the koji.hub.ConfigFile environment variable
-    in the httpd config. To override this (for example):
+    Options are read from one or more config files.
+    The config file location can be controlled by the koji.hub.ConfigFile and
+    koji.hub.ConfigDir environment variables in the httpd config.
+    To override this (for example):
 
       SetEnv koji.hub.ConfigFile /home/developer/koji/hub/hub.conf
-
-    Backwards compatibility:
-        - if ConfigFile is not set, opts are loaded from http config
-        - if ConfigFile is set, then the http config must not provide Koji options
-        - In a future version we will load the default hub config regardless
-        - all PythonOptions (except ConfigFile) are now deprecated and support for them
-          will disappear in a future version of Koji
     """
     # get our config file(s)
     cf = environ.get('koji.hub.ConfigFile', '/etc/koji-hub/hub.conf')
     cfdir = environ.get('koji.hub.ConfigDir', '/etc/koji-hub/hub.conf.d')
     config = koji.read_config_files([cfdir, (cf, True)], raw=True)
 
+    # load defined options from [hub] section
     opts = {}
     for name, dtype, default in config_map:
         key = ('hub', name)
@@ -566,15 +561,28 @@ def load_config(environ):
         opts['DBHost'] = opts['DBhost']
     if opts['RLIMIT_NOFILE'] is None:
         opts['RLIMIT_NOFILE'] = opts['RLIMIT_OFILE']
-    # load policies
-    # (only from config file)
+
+    # load policies from [policy] section
     if config and config.has_section('policy'):
         # for the moment, we simply transfer the policy conf to opts
+        # this gets parsed later in get_policy()
         opts['policy'] = dict(config.items('policy'))
     else:
         opts['policy'] = {}
     for pname, text in _default_policies.items():
         opts['policy'].setdefault(pname, text)
+
+    # load plugin.* sections
+    opts['plugin_conf'] = {}
+    for sect in config.sections():
+        if sect.startswith('plugin.'):
+            pname = sect[7:]
+            if not pname:
+                raise ValueError(f'Invalid section name {sect}')
+            # we don't know what options the plugin uses, so just read everything
+            pconf = {n: config.get(sect, n) for n in config.options(sect)}
+            opts['plugin_conf'][pname] = pconf
+
     # use configured KojiDir
     if opts.get('KojiDir') is not None:
         koji.BASEDIR = opts['KojiDir']
@@ -596,7 +604,11 @@ def load_plugins(opts):
     for name in opts['Plugins'].split():
         logger.info('Loading plugin: %s', name)
         try:
-            tracker.load(name)
+            plugin = tracker.load(name)
+            if hasattr(plugin, 'set_config'):
+                # pass in any plugin config we have
+                pconf = opts['plugin_conf'].get(name, {})
+                getattr(plugin, 'set_config')(pconf)
         except Exception:
             logger.error(''.join(traceback.format_exception(*sys.exc_info())))
             # make this non-fatal, but set ServerOffline

plugins/hub/message.py

@@ -0,0 +1,390 @@
+# Koji message plugin
+# Copyright (c) 2026 Red Hat, Inc.
+#
+# SPDX-License-Identifier: GPL-2.0-or-later
+
+import json
+import logging
+from uuid import uuid4
+
+import koji
+from koji.context import context
+from koji.plugin import callback, convert_datetime, export_in
+from kojihub.kojihub import get_build_type, convert_value
+from kojihub.db import QueryProcessor, InsertProcessor, DeleteProcessor, UpdateProcessor, db_lock
+
+
+'''
+This hub plugin captures key events using the callback mechanism and records
+messages in database. It also provides a simple api for a client to retrieve
+these messages.
+
+It does not send these messages to any external system directly. That is left
+for the client to implement.
+'''
+
+
+CONFIG = {
+    'string_limit': 0,  # no limit by default
+    'item_limit': 0,  # no limit by default
+    'batch_size': 100,
+    'max_age': 24,  # hours to keep old messages
+    'max_sent_age': None,  # hours to keep old messages after sending
+    'auto_clean': True,
+    'auto_clean_interval': 300,  # seconds
+}
+logger = logging.getLogger('koji.plugin.message')
+
+
+def set_config(pconf):
+    # called when loaded on startup
+    global CONFIG
+    config = {}
+    for name in pconf:
+        value = pconf[name]
+        if name == 'string_limit':
+            value = convert_value(value, int)
+        elif name == 'item_limit':
+            value = convert_value(value, int)
+        elif name == 'batch_size':
+            value = convert_value(value, int)
+        elif name == 'max_age':
+            value = convert_value(value, int)
+        elif name == 'max_sent_age':
+            value = convert_value(value, int)
+        elif name == 'auto_clean':
+            value = parse_bool_str(value)
+        elif name == 'auto_clean_interval':
+            value = convert_value(value, int)
+        else:
+            logger.warning(f'Unknown config setting {name} for message plugin')
+        config[name] = value
+    CONFIG.update(config)
+
+
+def parse_bool_str(value, default=False):
+    '''Helper function for boolean options'''
+    if not isinstance(value, str):
+        # shouldn't happen
+        raise koji.ParameterError(f'Invalid boolean value: {value!r}')
+    value = value.lower()
+    if value in ('yes', 'true', '1', 'y', 't'):
+        return True
+    if value in ('no', 'false', '0', 'n', 'f'):
+        return False
+    if value.isdigit():
+        return bool(int(value))
+    raise koji.ParameterError(f'Invalid boolean value: {value}')
+
+
+def json_serialize(o):
+    """JSON helper to encode otherwise unserializable data types"""
+    if isinstance(o, set):
+        return list(o)
+    logger.error("Not JSON serializable data: %s" % repr(o))
+    return {"error": "Can't serialize", "type": type(o).__name__}
+
+
+def store_msg(topic, props, data):
+    """Store message to db"""
+    data = data.copy()
+    data['uuid'] = uuid4().hex
+    try:
+        body = json.dumps(filter_body(data), default=json_serialize)
+    except OmittedValue:
+        logger.error("Message body for %s exceeds size filter limits, dropping message", topic)
+        return
+    logger.debug(f"Queuing message for topic {topic}, body size {len(body)}")
+
+    props = json.dumps(props)
+    insert = InsertProcessor(table='message_queue')
+    insert.set(topic=topic, props=props, body=body)
+    insert.execute()
+
+
+def filter_body(data):
+    # pre-encoded body data
+    opts = {}
+    if CONFIG['item_limit']:
+        opts['max_items'] = CONFIG['item_limit']
+    if CONFIG['string_limit']:
+        opts['max_len'] = CONFIG['string_limit']
+    if not opts:
+        # none of our filters were requested
+        return data
+    else:
+        return SizeFilter(data, **opts).walk()
+
+
+class OmittedValue(Exception):
+    """Raised when a field is omitted"""
+    ...
+    # TODO unify
+
+
+class SizeFilter:
+    # TODO unify
+
+    def __init__(self, data, max_len=0, max_items=0):
+        self.data = data
+        self.max_len = max_len
+        self.max_items = max_items
+
+    def walk(self):
+        return self._walk(self.data)
+
+    def _walk(self, value):
+        # recurse if needed
+        if isinstance(value, (tuple, list, set)):
+            # we treat these the same because we're converting to json later
+            if self.max_items and len(value) > self.max_items:
+                raise OmittedValue('too many items')
+            value = [self._walk(x) for x in value]
+            # note that we don't catch OmittedValue
+            # omitting a list entry omits the whole list
+        elif isinstance(value, dict):
+            if self.max_items and len(value) > self.max_items:
+                raise OmittedValue('too many items')
+            ret = {}
+            omitted = []
+            for k in value:
+                try:
+                    k = self._walk(k)
+                    v = self._walk(value[k])
+                except OmittedValue:
+                    # just omit this entry
+                    omitted.append(k)
+                    continue
+                ret[k] = v
+            if omitted:
+                ret['_OMITTED_KEYS'] = omitted
+            value = ret
+        elif isinstance(value, (str, bytes)):
+            if self.max_len and len(value) > self.max_len:
+                raise OmittedValue('string too long')
+        # else
+        return value
+
+
+@convert_datetime
+@callback('postPackageListChange')
+def prep_package_list_change(cbtype, *args, **kws):
+    topic = 'package.' + kws['action']
+    props = {'type': cbtype[4:],
+             'tag': kws['tag']['name'],
+             'package': kws['package']['name'],
+             'action': kws['action'],
+             'user': kws['user']['name']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postTaskStateChange')
+def prep_task_state_change(cbtype, *args, **kws):
+    if kws['attribute'] != 'state':
+        return
+    topic = 'task.' + kws['new'].lower()
+    props = {'type': cbtype[4:],
+             'id': kws['info']['id'],
+             'parent': kws['info']['parent'],
+             'method': kws['info']['method'],
+             'attribute': kws['attribute'],
+             'old': kws['old'],
+             'new': kws['new']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postBuildStateChange')
+def prep_build_state_change(cbtype, *args, **kws):
+    if kws['attribute'] != 'state':
+        return
+    old = kws['old']
+    if old is not None:
+        old = koji.BUILD_STATES[old]
+    new = koji.BUILD_STATES[kws['new']]
+    topic = 'build.' + new.lower()
+    kws['info'] = kws['info']
+    kws['btypes'] = get_build_type(kws['info'])
+    props = {'type': cbtype[4:],
+             'name': kws['info']['name'],
+             'version': kws['info']['version'],
+             'release': kws['info']['release'],
+             'attribute': kws['attribute'],
+             'old': old,
+             'new': new}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postImport')
+def prep_import(cbtype, *args, **kws):
+    kws['build'] = kws['build']
+    topic = 'import.' + kws['type']
+    props = {'type': cbtype[4:],
+             'importType': kws['type'],
+             'name': kws['build']['name'],
+             'version': kws['build']['version'],
+             'release': kws['build']['release']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postRPMSign')
+def prep_rpm_sign(cbtype, *args, **kws):
+    if not kws['sigkey']:
+        return
+    kws['build'] = kws['build']
+    topic = 'sign.rpm'
+    props = {'type': cbtype[4:],
+             'sigkey': kws['sigkey'],
+             'name': kws['build']['name'],
+             'version': kws['build']['version'],
+             'release': kws['build']['release'],
+             'rpm_name': kws['rpm']['name'],
+             'rpm_version': kws['rpm']['version'],
+             'rpm_release': kws['rpm']['release'],
+             'rpm_arch': kws['rpm']['arch']}
+    store_msg(topic, props, kws)
+
+
+def _prep_tag_msg(topic, cbtype, kws):
+    kws['build'] = kws['build']
+    props = {'type': cbtype[4:],
+             'tag': kws['tag']['name'],
+             'name': kws['build']['name'],
+             'version': kws['build']['version'],
+             'release': kws['build']['release'],
+             'user': kws['user']['name']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postTag')
+def prep_tag(cbtype, *args, **kws):
+    _prep_tag_msg('build.tag', cbtype, kws)
+
+
+@convert_datetime
+@callback('postUntag')
+def prep_untag(cbtype, *args, **kws):
+    _prep_tag_msg('build.untag', cbtype, kws)
+
+
+@convert_datetime
+@callback('postRepoInit')
+def prep_repo_init(cbtype, *args, **kws):
+    kws['task_id'] = kws.get('task_id')
+    topic = 'repo.init'
+    props = {'type': cbtype[4:],
+             'tag': kws['tag']['name'],
+             'repo_id': kws['repo_id'],
+             'task_id': kws['task_id']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postRepoDone')
+def prep_repo_done(cbtype, *args, **kws):
+    kws['task_id'] = kws.get('task_id')
+    topic = 'repo.done'
+    props = {'type': cbtype[4:],
+             'tag': kws['repo']['tag_name'],
+             'repo_id': kws['repo']['id'],
+             'task_id': kws['repo']['task_id'],
+             'expire': kws['expire']}
+    store_msg(topic, props, kws)
+
+
+@convert_datetime
+@callback('postBuildPromote')
+def prep_build_promote(cbtype, *args, **kws):
+    kws['build'] = kws['build']
+    topic = 'build.promote'
+    props = {'type': cbtype[4:],
+             'build_id': kws['build']['id'],
+             'name': kws['build']['name'],
+             'version': kws['build']['version'],
+             'release': kws['build']['release'],
+             'draft_release': kws['draft_release'],
+             'target_release': kws['target_release'],
+             'user': kws['user']['name']}
+    store_msg(topic, props, kws)
+
+
+@export_in('message')
+def poll(limit=None):
+    """Query pending messages"""
+    context.session.assertPerm('message')
+    if limit is None:
+        limit = CONFIG['batch_size']
+    else:
+        limit = convert_value(limit, cast=int)
+        if CONFIG['batch_size']:
+            limit = min(limit, CONFIG['batch_size'])
+
+    query = QueryProcessor(tables=('message_queue',),
+                           columns=('id', 'topic', 'props', 'body'),
+                           aliases=('id', 'topic', 'props', 'body'),
+                           clauses=['sent IS FALSE'],
+                           opts={'order': 'id', 'limit': limit})
+    return query.execute()
+
+
+@export_in('message')
+def ack(msg_ids):
+    """Acknowledge processed messages"""
+    context.session.assertPerm('message')
+    convert_value(msg_ids, cast=list, check_only=True)
+    msg_ids = [convert_value(_id, cast=int) for _id in msg_ids]
+    if not msg_ids:
+        raise koji.ParameterError('Empty message list')
+
+    if CONFIG['auto_clean']:
+        # check age of last acked message
+        query = QueryProcessor(tables=('message_queue',),
+                               columns=("date_part('epoch', now() - max(send_time))",),
+                               aliases=('age',),
+                               clauses=['sent IS TRUE'])
+        age = query.singleValue(strict=False)
+        if age is None or age > CONFIG['auto_clean_interval']:
+            clean()
+
+    # mark messages sent
+    update = UpdateProcessor('message_queue',
+                             clauses=['id IN %(ids)s', 'sent IS FALSE'],
+                             values={'ids': msg_ids},
+                             data={'sent': True},
+                             rawdata={'send_time': 'NOW()'})
+    update.execute()
+
+
+@export_in('message')
+def clean(force=False):
+    context.session.assertPerm('message')
+    force = convert_value(force, cast=bool)
+
+    if not db_lock('message-plugin-clean', wait=force):
+        # already running elsewhere
+        return False
+
+    logger.info('cleaning message queue')
+
+    # delete old messages
+    max_age = CONFIG['max_age']
+    delete = DeleteProcessor(
+        table='message_queue', values={'age': max_age},
+        clauses=["create_time < NOW() - (%(age)s::text || ' hours')::interval"])
+    delete.execute()
+
+    # sent messages can be deleted faster if configured
+    max_sent_age = CONFIG['max_sent_age']
+    if max_sent_age:
+        delete = DeleteProcessor(
+            table='message_queue', values={'age': max_sent_age},
+            clauses=["sent IS TRUE",
+                     "create_time < NOW() - (%(age)s::text || ' hours')::interval"])
+        delete.execute()
+
+
+# the end

schemas/schema-upgrade-1.36-1.37.sql

@@ -0,0 +1,21 @@
+-- upgrade script to migrate the Koji database schema
+-- from version 1.36 to 1.37
+
+BEGIN;
+
+-- Updates for the new message plugin
+CREATE TABLE message_queue (
+        id SERIAL PRIMARY KEY,
+        create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        send_time TIMESTAMPTZ,
+        sent BOOLEAN NOT NULL DEFAULT FALSE,
+        topic TEXT NOT NULL,
+        props JSONB NOT NULL,
+        body JSONB NOT NULL
+) WITHOUT OIDS;
+
+INSERT INTO locks(name) VALUES('message-plugin-clean');
+INSERT INTO permissions (name, description) VALUES ('message', 'Manage the message plugin queue.');
+
+
+COMMIT;

schemas/schema.sql

@@ -67,6 +67,8 @@ INSERT INTO permissions (name, description) VALUES ('target', 'Add, edit, and re
 INSERT INTO permissions (name, description) VALUES ('win-admin', 'The default hub policy rule for "vm" requires this permission to trigger Windows builds.');
 INSERT INTO permissions (name, description) VALUES ('win-import', 'Import win archives.');
 INSERT INTO permissions (name, description) VALUES ('draft-promoter', 'The permission required in the default "draft_promotion" hub policy rule to promote draft build.');
+INSERT INTO permissions (name, description) VALUES ('message', 'Manage the message plugin queue.');
+
 
 CREATE TABLE user_perms (
 	user_id INTEGER NOT NULL REFERENCES users(id),
@@ -1066,6 +1068,19 @@ CREATE TABLE proton_queue (
         body JSON NOT NULL
 ) WITHOUT OIDS;
 
+
+-- Queue for the message plugin
+CREATE TABLE message_queue (
+        id SERIAL PRIMARY KEY,
+        create_time TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+        send_time TIMESTAMPTZ,
+        sent BOOLEAN NOT NULL DEFAULT FALSE,
+        topic TEXT NOT NULL,
+        props JSONB NOT NULL,
+        body JSONB NOT NULL
+) WITHOUT OIDS;
+
+
 -- track checksum of rpms
 CREATE TABLE rpm_checksum (
         rpm_id INTEGER NOT NULL REFERENCES rpminfo(id),
@@ -1130,5 +1145,6 @@ CREATE TABLE locks (
 INSERT INTO locks(name) VALUES('protonmsg-plugin');
 INSERT INTO locks(name) VALUES('scheduler');
 INSERT INTO locks(name) VALUES('repo-queue');
+INSERT INTO locks(name) VALUES('message-plugin-clean');
 
 COMMIT WORK;

tests/test_hub/test_config.py

@@ -106,5 +106,27 @@ channel =
             name, dtype, default = row
             self.assertIn(dtype, ('integer', 'boolean', 'string'))
 
+    def test_plugin_conf(self):
+        config = '''
+[plugin.message]
+extra_limit = 1024
+max_age = 48
+'''
+        self.write_config_string(config)
+
+        kojixmlrpc.load_config(self.environ)
+
+    def test_bad_plugin_conf(self):
+        config = '''
+[plugin.]
+extra_limit = 1024
+max_age = 48
+'''
+        self.write_config_string(config)
+
+        with self.assertRaises(ValueError) as ex:
+            kojixmlrpc.load_config(self.environ)
+        assert str(ex.exception).startswith('Invalid section name')
+
 
 # the end

tests/test_plugins/test_message.py

@@ -0,0 +1,539 @@
+from __future__ import absolute_import
+
+import json
+import unittest
+from copy import deepcopy
+
+import koji
+import message  # our plugin
+
+try:
+    from unittest import mock
+except ImportError:
+    import mock
+
+
+IP = message.InsertProcessor
+QP = message.QueryProcessor
+UP = message.UpdateProcessor
+DP = message.DeleteProcessor
+
+
+class TestMessage(unittest.TestCase):
+
+    def setUp(self):
+        dup_conf = message.CONFIG.copy()
+        mock.patch('message.CONFIG', new=dup_conf).start()
+        mock.patch('message.logger').start()
+        self.get_build_type = mock.patch('message.get_build_type').start()
+        self.inserts = []
+        self.InsertProcessor = mock.patch('message.InsertProcessor',
+                                          side_effect=self.getInsert).start()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def getInsert(self, *args, **kwargs):
+        insert = IP(*args, **kwargs)
+        insert.execute = mock.MagicMock()
+        self.inserts.append(insert)
+        return insert
+
+    def assertMsg(self, topic, body=None, **kws):
+        self.assertEqual(len(self.inserts), 1)
+        insert = self.inserts[0]
+        self.assertEqual(insert.table, 'message_queue')
+        self.assertEqual(insert.data['topic'], topic)
+        if kws:
+            msg_props = json.loads(insert.data['props'])
+            self.assertDictEqual(msg_props, kws)
+        if body is not None:
+            # if isinstance(body, str):
+            #     body = json.loads(body)
+            msg_body = json.loads(insert.data['body'])
+            if 'uuid' in msg_body:
+                # ignore our random uuid
+                del msg_body['uuid']
+            self.assertDictEqual(msg_body, body)
+
+    def test_prep_package_list_change_add(self):
+        message.prep_package_list_change('postPackageListChange',
+                                         action='add', tag={'name': 'test-tag'},
+                                         package={'name': 'test-pkg'},
+                                         owner=1,
+                                         block=False, extra_arches='i386 x86_64',
+                                         force=False, update=False,
+                                         user={'name': 'username'})
+        self.assertMsg('package.add', type='PackageListChange', tag='test-tag',
+                       package='test-pkg', action='add', user='username')
+
+    def test_prep_package_list_change_update(self):
+        message.prep_package_list_change('postPackageListChange',
+                                         action='update', tag={'name': 'test-tag'},
+                                         package={'name': 'test-pkg'},
+                                         owner=1,
+                                         block=False, extra_arches='i386 x86_64',
+                                         force=False, update=False,
+                                         user={'name': 'username'})
+        self.assertMsg('package.update', type='PackageListChange', tag='test-tag',
+                       package='test-pkg', action='update', user='username')
+
+    def test_prep_package_list_change_block(self):
+        message.prep_package_list_change('postPackageListChange',
+                                         action='block', tag={'name': 'test-tag'},
+                                         package={'name': 'test-pkg'},
+                                         owner=1,
+                                         block=False, extra_arches='i386 x86_64',
+                                         force=False, update=False,
+                                         user={'name': 'username'})
+        self.assertMsg('package.block', type='PackageListChange', tag='test-tag',
+                       package='test-pkg', action='block', user='username')
+
+    def test_prep_package_list_change_unblock(self):
+        message.prep_package_list_change('postPackageListChange',
+                                         action='unblock', tag={'name': 'test-tag'},
+                                         package={'name': 'test-pkg'},
+                                         user={'name': 'username'})
+        self.assertMsg('package.unblock', type='PackageListChange', tag='test-tag',
+                       package='test-pkg', action='unblock', user='username')
+
+    def test_prep_package_list_change_remove(self):
+        message.prep_package_list_change('postPackageListChange',
+                                         action='remove', tag={'name': 'test-tag'},
+                                         package={'name': 'test-pkg'},
+                                         user={'name': 'username'})
+        self.assertMsg('package.remove', type='PackageListChange', tag='test-tag',
+                       package='test-pkg', action='remove', user='username')
+
+    def test_prep_task_state_change(self):
+        info = {'id': 5678,
+                'parent': 1234,
+                'method': 'build'}
+        message.prep_task_state_change('postTaskStateChange',
+                                       info=info, attribute='weight',
+                                       old=2.0, new=3.5)
+        # no messages should be created for callbacks where attribute != state
+        message.prep_task_state_change('postTaskStateChange',
+                                       info=info, attribute='state',
+                                       old='FREE', new='OPEN')
+        self.assertMsg('task.open', type='TaskStateChange',
+                       attribute='state', old='FREE', new='OPEN',
+                       **info)
+
+    def test_prep_build_state_change(self):
+        info = {'name': 'test-pkg',
+                'version': '1.0',
+                'release': '1',
+                'build_id': 1}
+        assert_info = {'name': 'test-pkg',
+                       'version': '1.0',
+                       'release': '1'}
+        msg_body = {
+            'info': info.copy(),
+            'btypes': {'image': {'build_id': 1}},
+            "attribute": "state",
+            "old": 0,
+            "new": 1,
+        }
+        self.get_build_type.return_value = {'image': {'build_id': 1}}
+        message.prep_build_state_change('postBuildStateChange',
+                                        info=info, attribute='volume_id',
+                                        old=0, new=1)
+        # no messages should be created for callbacks where attribute != state
+        self.get_build_type.return_value = {'image': {'build_id': 1}}
+        message.prep_build_state_change('postBuildStateChange',
+                                        info=info, attribute='state',
+                                        old=0, new=1)
+        self.assertMsg('build.complete', type='BuildStateChange',
+                       attribute='state', old='BUILDING', new='COMPLETE',
+                       body=msg_body, **assert_info)
+
+    def test_prep_build_state_change_with_empty_build_type(self):
+        info = {'name': 'test-pkg',
+                'version': '1.0',
+                'release': '1',
+                'build_id': 1}
+        assert_info = {'name': 'test-pkg',
+                       'version': '1.0',
+                       'release': '1'}
+        msg_body = {
+            'info': info.copy(),
+            'btypes': {},
+            'attribute': 'state',
+            'old': 0,
+            'new': 1,
+        }
+        self.get_build_type.return_value = {}
+        message.prep_build_state_change('postBuildStateChange',
+                                        info=info, attribute='volume_id',
+                                        old=0, new=1)
+        # no messages should be created for callbacks where attribute != state
+        self.get_build_type.return_value = {}
+        message.prep_build_state_change('postBuildStateChange',
+                                        info=info, attribute='state',
+                                        old=0, new=1)
+        self.assertMsg('build.complete', type='BuildStateChange',
+                       attribute='state', old='BUILDING', new='COMPLETE',
+                       body=msg_body, **assert_info)
+
+    def test_prep_import(self):
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        message.prep_import('postImport', type='build', build=build)
+        self.assertMsg('import.build', type='Import', importType='build',
+                       **build)
+
+    def test_prep_rpm_sign(self):
+        build = {'name': 'test-pkg',
+                 'version': '1.0',
+                 'release': '1'}
+        rpm = {'name': 'test-pkg-subpkg',
+               'version': '2.0',
+               'release': '2',
+               'arch': 'x86_64'}
+        sigkey = 'a1b2c3d4'
+        message.prep_rpm_sign('postRPMSign', sigkey=sigkey, sighash='fedcba9876543210',
+                              build=build, rpm=rpm)
+        self.assertMsg('sign.rpm', type='RPMSign', sigkey=sigkey, rpm_name=rpm['name'],
+                       rpm_version=rpm['version'], rpm_release=rpm['release'],
+                       rpm_arch='x86_64',
+                       **build)
+
+    def test_prep_rpm_sign_no_sigkey(self):
+        build = {'name': 'test-pkg',
+                 'version': '1.0',
+                 'release': '1'}
+        rpm = {'name': 'test-pkg-subpkg',
+               'version': '2.0',
+               'release': '2',
+               'arch': 'x86_64'}
+        sigkey = ''
+        message.prep_rpm_sign('postRPMSign', sigkey=sigkey, sighash='fedcba9876543210',
+                              build=build, rpm=rpm)
+        # the plugin does not issue messages for blank sigkey
+        self.assertEqual(len(self.inserts), 0)
+
+    def test_prep_tag(self):
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        message.prep_tag('postTag', tag={'name': 'test-tag'},
+                         build=build, user={'name': 'test-user'})
+        self.assertMsg('build.tag', type='Tag', tag='test-tag',
+                       user='test-user', **build)
+
+    def test_prep_untag(self):
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        message.prep_untag('postUntag', tag={'name': 'test-tag'},
+                           build=build, user={'name': 'test-user'})
+        self.assertMsg('build.untag', type='Untag', tag='test-tag',
+                       user='test-user', **build)
+
+    def test_prep_repo_init(self):
+        message.prep_repo_init('postRepoInit',
+                               tag={'name': 'test-tag', 'arches': set(['x86_64', 'i386'])},
+                               repo_id=1234, task_id=25)
+        self.assertMsg('repo.init', type='RepoInit', tag='test-tag', repo_id=1234, task_id=25)
+
+    def test_prep_repo_done(self):
+        message.prep_repo_done('postRepoDone',
+                               repo={'tag_name': 'test-tag', 'id': 1234, 'task_id': 25},
+                               expire=False)
+        self.assertMsg('repo.done', type='RepoDone', tag='test-tag', repo_id=1234,
+                       task_id=25, expire=False)
+
+    def test_prep_build_promote(self):
+        build = {'id': 12345, 'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        message.prep_build_promote('postBuildPromote',
+                                   draft_release='1,draft_12345',
+                                   target_release='1',
+                                   build=build,
+                                   user={'name': 'test-user'})
+        expect = {
+            'build_id': build['id'],
+            'name': build['name'],
+            'version': build['version'],
+            'release': build['release'],
+            'draft_release': '1,draft_12345',
+            'target_release': build['release'],
+            'type': 'BuildPromote',
+            'user': 'test-user',
+        }
+        self.assertMsg('build.promote', **expect)
+
+    def test_string_limit(self):
+        message.set_config({'string_limit': 1024})
+        metadata = {'bad': 'too long ' * 1024, 'good': 'hello world'}
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        params = {'type': 'cg',
+                  'metadata': metadata,
+                  'build': build,
+                  'directory': '/tmp/whatever'}
+        message.prep_import('postImport', **params)
+
+        expect = deepcopy(params)
+        del expect['metadata']['bad']
+        expect['metadata']['_OMITTED_KEYS'] = ['bad']
+        build_kw = build.copy()
+        self.assertMsg('import.cg', body=expect, type='Import',
+                       importType='cg', **build_kw)
+
+    def test_item_limit(self):
+        message.set_config({'item_limit': 10})
+        metadata = {'bad': [['too many'] * 12], 'good': 'hello world'}
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        params = {'type': 'cg',
+                  'metadata': metadata,
+                  'build': build,
+                  'directory': '/tmp/whatever'}
+        message.prep_import('postImport', **params)
+
+        expect = deepcopy(params)
+        del expect['metadata']['bad']
+        expect['metadata']['_OMITTED_KEYS'] = ['bad']
+        build_kw = build.copy()
+        self.assertMsg('import.cg', body=expect, type='Import',
+                       importType='cg', **build_kw)
+
+    def test_item_limit_dict(self):
+        message.set_config({'item_limit': 10})
+        bad = {f'key_{i}': i for i in range(12)}
+        metadata = {'bad': bad, 'good': 'hello world'}
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        params = {'type': 'cg',
+                  'metadata': metadata,
+                  'build': build,
+                  'directory': '/tmp/whatever'}
+        message.prep_import('postImport', **params)
+
+        expect = deepcopy(params)
+        del expect['metadata']['bad']
+        expect['metadata']['_OMITTED_KEYS'] = ['bad']
+        build_kw = build.copy()
+        self.assertMsg('import.cg', body=expect, type='Import',
+                       importType='cg', **build_kw)
+
+    def test_item_limit_toplevel(self):
+        message.set_config({'item_limit': 3})
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1'}
+        params = {'type': 'cg',
+                  'metadata': {'hello': 'world'},
+                  'build': build,
+                  'directory': '/tmp/whatever'}
+        message.prep_import('postImport', **params)
+        self.assertEqual(len(self.inserts), 0)
+        message.logger.error.assert_called_once()
+        self.assertIn('exceeds size filter', message.logger.error.call_args[0][0])
+
+    def test_invalid_type(self):
+        # a non-serializable type should be filtered
+        build = {'name': 'test-pkg', 'version': '1.0', 'release': '1',
+                 'bad': b'bytes are not serializable'}
+        build['nvr'] = '%(name)s-%(version)s-%(release)s' % build
+        params = {'tag': {'name': 'test-tag'},
+                  'build': build,
+                  'user': {'name': 'test-user'}}
+        message.prep_tag('postTag', **params)
+
+        body = deepcopy(params)
+        del body['build']['bad']
+        body['build']['bad'] = {'error': "Can't serialize", 'type': "bytes"}
+        build_kw = build.copy()
+        del build_kw['bad']
+        del build_kw['nvr']
+        self.assertMsg('build.tag', body=body, type='Tag', tag='test-tag',
+                       user='test-user', **build_kw)
+
+
+class TestConfig(unittest.TestCase):
+
+    def setUp(self):
+        self.orig = message.CONFIG.copy()
+        dup = self.orig.copy()
+        mock.patch('message.CONFIG', new=dup).start()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def test_blank_config(self):
+        conf = {}
+        message.set_config(conf)
+        self.assertEqual(message.CONFIG, self.orig)
+
+    def test_nonblank_config(self):
+        conf = {'string_limit': '8192', 'batch_size': '42', 'max_age': '23'}
+        expect = self.orig.copy()
+        expect.update({'string_limit': 8192, 'batch_size': 42, 'max_age': 23})
+        message.set_config(conf)
+        self.assertEqual(message.CONFIG, expect)
+
+    def test_unknown_config(self):
+        # an unknown config option should be treated as a string
+        conf = {'batch_size': '137', 'other_option': 'hello'}
+        expect = self.orig.copy()
+        expect['batch_size'] = 137
+        expect['other_option'] = 'hello'
+        message.set_config(conf)
+        self.assertEqual(message.CONFIG, expect)
+
+
+class TestExports(unittest.TestCase):
+
+    def setUp(self):
+        dup_conf = message.CONFIG.copy()
+        mock.patch('message.CONFIG', new=dup_conf).start()
+        self.context = mock.patch('message.context').start()
+        self.queries = []
+        self.context.session.assertPerm = mock.MagicMock()
+        self.QueryProcessor = mock.patch('message.QueryProcessor',
+                                         side_effect=self.getQuery).start()
+        self.updates = []
+        self.UpdateProcessor = mock.patch('message.UpdateProcessor',
+                                          side_effect=self.getUpdate).start()
+        self.deletes = []
+        self.DeleteProcessor = mock.patch('message.DeleteProcessor',
+                                          side_effect=self.getDelete).start()
+        self.db_lock = mock.patch('message.db_lock').start()
+        self.query_execute = mock.MagicMock()
+        self.singleValue = mock.MagicMock()
+
+    def tearDown(self):
+        mock.patch.stopall()
+
+    def getQuery(self, *args, **kwargs):
+        query = QP(*args, **kwargs)
+        query.execute = self.query_execute
+        query.singleValue = self.singleValue
+        self.queries.append(query)
+        return query
+
+    def getUpdate(self, *args, **kwargs):
+        update = UP(*args, **kwargs)
+        update.execute = mock.MagicMock()
+        self.updates.append(update)
+        return update
+
+    def getDelete(self, *args, **kwargs):
+        delete = DP(*args, **kwargs)
+        delete.execute = mock.MagicMock()
+        self.deletes.append(delete)
+        return delete
+
+    def test_poll(self):
+        message.poll()
+
+        self.context.session.assertPerm.assert_called_once_with('message')
+        assert len(self.queries) == 1
+        query = self.queries[0]
+        assert query.tables == ('message_queue',)
+
+    def test_poll_limit(self):
+        message.poll(5)
+
+        self.context.session.assertPerm.assert_called_once_with('message')
+        assert len(self.queries) == 1
+        query = self.queries[0]
+        assert query.tables == ('message_queue',)
+        assert query.opts['limit'] == 5
+
+        message.set_config({'batch_size': 99})
+        message.poll(5000)  # over our max
+        assert len(self.queries) == 2
+        query = self.queries[1]
+        assert query.opts['limit'] == 99
+
+    def test_poll_invalid_limit(self):
+        with self.assertRaises(koji.ParameterError):
+            message.poll([5])
+        with self.assertRaises(koji.ParameterError):
+            message.poll('hello')
+
+    def test_ack_noclean(self):
+        message.set_config({'auto_clean': 'false'})
+        msg_ids = [4, 5, 6]
+        message.ack(msg_ids)
+
+        self.context.session.assertPerm.assert_called_once_with('message')
+        assert len(self.queries) == 0
+        assert len(self.updates) == 1
+        assert len(self.deletes) == 0
+        update = self.updates[0]
+        assert update.table == 'message_queue'
+
+    def test_auto_clean_skipped(self):
+        # auto clean is on by default
+        message.set_config({'auto_clean_interval': '60'})
+        self.singleValue.return_value = 55  # less than interval
+        msg_ids = [99]
+        message.ack(msg_ids)
+
+        self.context.session.assertPerm.assert_called_once_with('message')
+        assert len(self.queries) == 1
+        assert len(self.updates) == 1
+        assert len(self.deletes) == 0
+        update = self.updates[0]
+        assert update.table == 'message_queue'
+
+    def test_auto_clean_runs(self):
+        # auto clean is on by default
+        message.set_config({'auto_clean_interval': '60'})
+        self.singleValue.return_value = 65  # more than interval
+        msg_ids = [99]
+        message.ack(msg_ids)
+
+        self.context.session.assertPerm.assert_called_with('message')
+        assert len(self.queries) == 1
+        assert len(self.updates) == 1
+        assert len(self.deletes) == 1  # no max_sent_age by default
+        update = self.updates[0]
+        assert update.table == 'message_queue'
+
+    def test_invalid_ack(self):
+        msg_ids = 'hello'
+        with self.assertRaises(koji.ParameterError):
+            message.ack(msg_ids)
+
+        msg_ids = ['hello']
+        with self.assertRaises(koji.ParameterError):
+            message.ack(msg_ids)
+
+        msg_ids = []
+        with self.assertRaises(koji.ParameterError):
+            message.ack(msg_ids)
+
+    def test_max_sent(self):
+        message.set_config({'max_sent_age': 2, 'max_age': 48})
+        message.clean()
+
+        self.context.session.assertPerm.assert_called_once_with('message')
+        assert len(self.queries) == 0
+        assert len(self.updates) == 0
+        assert len(self.deletes) == 2
+        assert self.deletes[0].table == 'message_queue'
+        assert self.deletes[0].values['age'] == 48
+        assert self.deletes[1].table == 'message_queue'
+        assert self.deletes[1].values['age'] == 2
+
+    def test_clean_lock(self):
+        self.db_lock.return_value = False
+        message.clean()
+        assert len(self.deletes) == 0
+
+    def test_bool_opts(self):
+        for val in ('yes', 'True', '1', 'y', 'T', '100'):
+            assert message.parse_bool_str(val) is True
+        for val in ('NO', 'false', '0', 'n', 'f'):
+            assert message.parse_bool_str(val) is False
+
+        # other types accepted if not strict, otherwise error
+        for val in (1, True, [1]):
+            with self.assertRaises(koji.ParameterError):
+                message.parse_bool_str(val)
+        for val in (0, False, [], {}):
+            with self.assertRaises(koji.ParameterError):
+                message.parse_bool_str(val)
+
+        # unknown strings accepted if not strict, otherwise error
+        for val in ('junk', 'not-a-bool', 'invalid', ''):
+            with self.assertRaises(koji.ParameterError):
+                message.parse_bool_str(val)
+
+
+# the end