erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

component pubsub: Implement "max_items":

Browse source 
A new `max_items` field has been added to `PubsubNode`. `0` means `max`. If any >0 value
is used, items will be automatically deleted by `cache_pubsub_items`. Items are currently
discarded (i.e. no notification is sent), this is the same behaviour as in legacy Libervia
Pubsub (formerly "SàT Pubsub").

rel 476
This commit is contained in:
Goffi 2026-04-11 12:47:40 +02:00
commit da2c2c49ea
4 changed files with 122 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

18
libervia/backend/memory/migration/versions/f9ce59cfc13f_add_max_items_to_pubsubnode.py
View file

@ -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")

34
libervia/backend/memory/sqla.py
View file

@ -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,

1
libervia/backend/memory/sqla_mapping.py
View file

@ -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

86
libervia/backend/plugins/plugin_comp_pubsub/__init__.py
View file

@ -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: