diff --git a/libervia/backend/memory/migration/versions/f9ce59cfc13f_add_max_items_to_pubsubnode.py b/libervia/backend/memory/migration/versions/f9ce59cfc13f_add_max_items_to_pubsubnode.py index ea7d33e7..d667ddb5 100644 --- a/libervia/backend/memory/migration/versions/f9ce59cfc13f_add_max_items_to_pubsubnode.py +++ b/libervia/backend/memory/migration/versions/f9ce59cfc13f_add_max_items_to_pubsubnode.py @@ -5,22 +5,26 @@ Revises: ece2a6d7e4bb Create Date: 2026-04-11 12:29:56.062028 """ + from alembic import op import sqlalchemy as sa - # revision identifiers, used by Alembic. -revision = 'f9ce59cfc13f' -down_revision = 'ece2a6d7e4bb' +revision = "f9ce59cfc13f" +down_revision = "ece2a6d7e4bb" branch_labels = None depends_on = None def upgrade(): - with op.batch_alter_table('pubsub_nodes', schema=None) as batch_op: - batch_op.add_column(sa.Column('max_items', sa.Integer(), server_default=sa.text('0'), nullable=False)) + with op.batch_alter_table("pubsub_nodes", schema=None) as batch_op: + batch_op.add_column( + sa.Column( + "max_items", sa.Integer(), server_default=sa.text("0"), nullable=False + ) + ) def downgrade(): - with op.batch_alter_table('pubsub_nodes', schema=None) as batch_op: - batch_op.drop_column('max_items') + with op.batch_alter_table("pubsub_nodes", schema=None) as batch_op: + batch_op.drop_column("max_items") diff --git a/libervia/backend/memory/sqla.py b/libervia/backend/memory/sqla.py index 8b058111..6ee46f59 100644 --- a/libervia/backend/memory/sqla.py +++ b/libervia/backend/memory/sqla.py @@ -1443,6 +1443,7 @@ class Storage: subscriptions: list[PubsubSub] | None = None, parent_node: PubsubNode | None = None, linked_node: PubsubNode | None = None, + max_items: int = 0, ) -> PubsubNode: node = PubsubNode( profile_id=self.profiles[client.profile], @@ -1460,6 +1461,7 @@ class Storage: subscriptions=subscriptions or [], parent_node=parent_node, linked_node=linked_node, + max_items=max_items, ) try: async with self.session() as session: @@ -1533,7 +1535,9 @@ class Storage: items: list[domish.Element], parsed_items: list[dict] | None = None, ) -> None: - """Add items to database, using an upsert taking care of "updated" field + """Add items to database, using an upsert taking care of "updated" field. + + Enforces max_items limit by evicting oldest items if necessary. @param client: the client instance @param node: the pubsub node @@ -1567,6 +1571,20 @@ class Storage: ) ) await session.execute(stmt) + + # Enforce max_items limit if set + if node.max_items and node.max_items > 0: + subquery = ( + select(PubsubItem.id) + .filter_by(node_id=node.id) + .order_by(PubsubItem.updated.desc(), PubsubItem.id.desc()) + .offset(node.max_items) + ).subquery() + delete_stmt = delete(PubsubItem).filter( + PubsubItem.id.in_(select(subquery.c.id)) + ) + await session.execute(delete_stmt) + await session.commit() @aio @@ -1595,6 +1613,20 @@ class Storage: await session.execute(stmt) await session.commit() + @aio + async def get_pubsub_item_count(self, node: PubsubNode) -> int: + """Count the number of items in a node. + + @param node: the pubsub node + @return: the number of items in the node + """ + async with self.session() as session: + async with session.begin(): + result = await session.execute( + select(func.count(PubsubItem.id)).filter_by(node_id=node.id) + ) + return result.scalar() or 0 + @aio async def purge_pubsub_items( self, diff --git a/libervia/backend/memory/sqla_mapping.py b/libervia/backend/memory/sqla_mapping.py index 5e7c6d84..669681e3 100644 --- a/libervia/backend/memory/sqla_mapping.py +++ b/libervia/backend/memory/sqla_mapping.py @@ -651,6 +651,7 @@ class PubsubNode(Base): type_ = Column(Text, name="type", nullable=True) subtype = Column(Text, nullable=True) extra = Column(JSON) + max_items = Column(Integer, nullable=False, server_default=text("0")) linked_node_id = Column( Integer, ForeignKey("pubsub_nodes.id", ondelete="CASCADE"), nullable=True diff --git a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py index 65ebb9b8..ec85d004 100644 --- a/libervia/backend/plugins/plugin_comp_pubsub/__init__.py +++ b/libervia/backend/plugins/plugin_comp_pubsub/__init__.py @@ -142,6 +142,11 @@ class PubsubNodeConfig: "label": label, "options": {value.value: None for value in enum_type}, } + case "max_items": + definitions[pubsub_var] = { + "type": "text-single", + "label": label, + } case _: definitions[pubsub_var] = { "type": "text-single", @@ -168,6 +173,7 @@ class PubsubResource(pubsub.PubSubResource): "delete-nodes", "item-ids", "meta-data", + "multi-items", "publish", "purge-nodes", "retract-items", @@ -292,7 +298,8 @@ class PubsubResource(pubsub.PubSubResource): client, client.jid, request.nodeIdentifier, - with_affiliations=True, with_parent=True, + with_affiliations=True, + with_parent=True, with_subscriptions=True, ) @@ -574,6 +581,22 @@ class PubsubResource(pubsub.PubSubResource): except ValueError: pass + # Handle max_items: "max" or any value <= 0 means unlimited (stored as 0), + # otherwise parse as int + max_items_value = request.options.get("pubsub#max_items") + if max_items_value is not None: + if max_items_value.lower() == "max": + config.max_items = "max" + else: + try: + parsed_max_items = int(max_items_value) + # Convert any value <= 0 to "max" + config.max_items = ( + "max" if parsed_max_items <= 0 else parsed_max_items + ) + except ValueError: + pass + # Store relationship fields and normalize empty strings to None config.parent = request.options.get(PARENT_VAR) or None config.link = request.options.get(LINK_VAR) or None @@ -584,6 +607,9 @@ class PubsubResource(pubsub.PubSubResource): ) try: + # "max" becomes 0 (unlimited), int stays as-is + db_max_items = 0 if config.max_items == "max" else config.max_items + node = await G.storage.set_pubsub_node( client=client, service=client.jid, @@ -591,6 +617,7 @@ class PubsubResource(pubsub.PubSubResource): access_model=config.access_model, publish_model=config.publish_model, affiliations=[owner_affiliation], + max_items=db_max_items, ) except exceptions.ConflictError: raise error.StanzaError("conflict", text=f"node {node_id!r} already exists.") @@ -680,7 +707,8 @@ class PubsubResource(pubsub.PubSubResource): client, client.jid, request.nodeIdentifier, - with_affiliations=True, with_parent=True, + with_affiliations=True, + with_parent=True, with_link=True, ) @@ -716,11 +744,10 @@ class PubsubResource(pubsub.PubSubResource): if node.publish_model: config_values["pubsub#publish_model"] = str(node.publish_model) - # TODO: Add max_items when it's stored in the database - # if node.max_items is not None: - # config_values["pubsub#max_items"] = ( - # "max" if node.max_items == -1 else str(node.max_items) - # ) + if node.max_items is not None: + config_values["pubsub#max_items"] = ( + "max" if node.max_items == 0 else str(node.max_items) + ) if node.linked_node is not None: config_values[LINK_VAR] = node.linked_node.name @@ -750,7 +777,8 @@ class PubsubResource(pubsub.PubSubResource): client, client.jid, request.nodeIdentifier, - with_affiliations=True, with_parent=True, + with_affiliations=True, + with_parent=True, with_link=True, ) @@ -783,6 +811,7 @@ class PubsubResource(pubsub.PubSubResource): return config = PubsubNodeConfig() + update_values: dict = {} access_model_value = request.options.get(XEP_0060.OPT_ACCESS_MODEL) if access_model_value: @@ -804,6 +833,46 @@ class PubsubResource(pubsub.PubSubResource): text=f"Invalid publish_model value: {publish_model_value!r}", ) + # Max_items + max_items_value = request.options.get("pubsub#max_items") + if max_items_value is not None: + max_items_value = max_items_value.strip().lower() + if max_items_value == "max": + # "max" means unlimited, stored as 0 in database + db_max_items = 0 + elif max_items_value == "": + # Empty value, we ignore + db_max_items = None + else: + try: + db_max_items = int(max_items_value) + # Convert any value <= 0 to "max" (0) + if db_max_items <= 0: + db_max_items = 0 + except ValueError: + raise error.StanzaError( + "not-acceptable", + text=f"Invalid max_items value: {max_items_value!r}", + ) + + # max_items can't be lower than current item count. This is not a standard + # validation, but it may avoid catastrophic accidental item deletion. + if db_max_items is not None and db_max_items > 0: + items_count = await G.storage.get_pubsub_item_count(node) + if db_max_items < items_count: + raise error.StanzaError( + "not-acceptable", + text=( + f"max_items can't be set to {db_max_items} which is lower" + f" than the current number of items ({items_count}). Please" + " increase max_items or delete items from this node first." + ), + ) + + # Only update if value changed + if db_max_items is not None and db_max_items != node.max_items: + update_values["max_items"] = db_max_items + title_value = request.options.get("pubsub#title") description_value = request.options.get("pubsub#description") @@ -816,7 +885,6 @@ class PubsubResource(pubsub.PubSubResource): extra["description"] = description_value update_extra = True - update_values = {} if config.access_model != node.access_model: update_values["access_model"] = config.access_model if config.publish_model != node.publish_model: