Initial commit

.env.example

···
+# Set this to the hostname that you intend to run the service at
+HOSTNAME='feed.example.com'
+
+# You can obtain it by publishing of feed (run publish_feed.py)
+FEED_URI='at://did:plc:abcde.../app.bsky.feed.generator/example-fid-name...'
+
+# Your handle name
+HANDLE=''
+
+# Your app password
+PASSWORD=''
+
+# A short name for the record that will show in urls
+# Lowercase with no spaces.
+# Ex: whats-hot
+RECORD_NAME=''
+
+# A display name for your feed
+# Ex: What's Hot
+DISPLAY_NAME=''
+
+# (Optional) A description of your feed
+# Ex: Top trending content from the whole network
+DESCRIPTION='powered by The AT Protocol SDK for Python'
+
+# (Optional) The path to an image to be used as your feed's avatar
+# Ex: ./path/to/avatar.jpeg
+AVATAR_PATH=''
+
+# (Optional). Only use this if you want a service did different from did:web
+SERVICE_DID=''
+
+# (Optional). If your feed accepts interactions from clients
+ACCEPTS_INTERACTIONS='false'
+
+# (Optional). If your feed is a video feed
+IS_VIDEO_FEED='false'
+
+# (Optional). Ignore reply posts
+#IGNORE_REPLY_POSTS='true'
+
+# (Optional). Ignore posts with a created_at timestamp older than 1 day
+# to avoid including archived posts from X/Twitter
+#IGNORE_OLD_POSTS='true'

.flaskenv

···
+FLASK_APP=server.app
+FLASK_RUN_PORT=8000
+FLASK_RUN_HOST=127.0.0.1

.github/FUNDING.yml

···
+github: MarshalX

.gitignore

···
+.DS_Store
+.idea
+*.iml
+.env
+*.db
+
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+bin/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+pyvenv.cfg
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/

.vscode/settings.json

···
+{
+    "cSpell.words": [
+        "ATPROTO",
+        "bluesky",
+        "bsky",
+        "dotenv"
+    ]
+}

LICENSE

···
+MIT License
+
+Copyright (c) 2023 Ilya Siamionau
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

···
+# ATProto Feed Generator powered by [The AT Protocol SDK for Python](https://github.com/MarshalX/atproto)
+
+> Feed Generators are services that provide custom algorithms to users through the AT Protocol.
+
+Official overview (read it first): https://github.com/bluesky-social/feed-generator#overview
+
+## Getting Started
+
+We've set up this simple server with SQLite to store and query data. Feel free to switch this out for whichever database you prefer.
+
+Next, you will need to do two things:
+
+1. Implement filtering logic in `server/data_filter.py`.
+2. Copy `.env.example` to `.env`
+3. Optionally implement custom feed generation logic in `server/algos`.
+
+We've taken care of setting this server up with a did:web. However, you're free to switch this out for did:plc if you like - you may want to if you expect this Feed Generator to be long-standing and possibly migrating domains.
+
+## Publishing your feed
+
+To publish your feed, simply run `python publish_feed.py`.
+
+To update your feed's display data (name, avatar, description, etc.), just update the relevant variables in `.env` and re-run the script.
+
+After successfully running the script, you should be able to see your feed from within the app, as well as share it by embedding a link in a post (similar to a quote post).
+
+## Running the Server
+
+Install Python 3.7+.
+
+Run `setupvenv.sh` to setup a virtual environment and install the dependencies:
+
+```shell
+./setupvenv.sh
+```
+
+**Note**: To get value for `FEED_URI` you need to publish the feed first
+
+To run a development Flask server:
+
+```shell
+flask run
+```
+
+**Warning** The Flask development server is not designed for production use. In production, you should use production WSGI server such as [`waitress`](https://flask.palletsprojects.com/en/stable/deploying/waitress/) behind a reverse proxy such as NGINX instead.
+
+```shell
+pip install waitress
+waitress-serve --listen=127.0.0.1:8080 server.app:app
+```
+
+To run a development server with debugging:
+
+```shell
+flask --debug run
+```
+
+**Note**: Duplication of data stream instances in debug mode is fine.
+
+**Warning**: If you want to run server in many workers, you should run Data Stream (Firehose) separately.
+
+### Endpoints
+
+- `/.well-known/did.json`
+- `/xrpc/app.bsky.feed.describeFeedGenerator`
+- `/xrpc/app.bsky.feed.getFeedSkeleton`
+
+## License
+
+MIT

publish_feed.py

···
+#!/usr/bin/env python3
+# YOU MUST INSTALL ATPROTO SDK
+# pip3 install atproto
+
+import os
+
+from dotenv import load_dotenv
+from atproto import Client, models
+
+load_dotenv()
+
+def _get_bool_env_var(value: str) -> bool:
+    # Helper function to convert string to bool
+
+    if value is None:
+        return False
+
+    normalized_value = value.strip().lower()
+    if normalized_value in {'1', 'true', 't', 'yes', 'y'}:
+        return True
+
+    return False
+
+
+# YOUR bluesky handle
+# Ex: user.bsky.social
+HANDLE: str = os.environ.get('HANDLE')
+
+# YOUR bluesky password, or preferably an App Password (found in your client settings)
+# Ex: abcd-1234-efgh-5678
+PASSWORD: str = os.environ.get('PASSWORD')
+
+# The hostname of the server where feed server will be hosted
+# Ex: feed.bsky.dev
+HOSTNAME: str = os.environ.get('HOSTNAME')
+
+# A short name for the record that will show in urls
+# Lowercase with no spaces.
+# Ex: whats-hot
+RECORD_NAME: str = os.environ.get('RECORD_NAME')
+
+# A display name for your feed
+# Ex: What's Hot
+DISPLAY_NAME: str = os.environ.get('DISPLAY_NAME')
+
+# (Optional) A description of your feed
+# Ex: Top trending content from the whole network
+DESCRIPTION: str = os.environ.get('DESCRIPTION')
+
+# (Optional) The path to an image to be used as your feed's avatar
+# Ex: ./path/to/avatar.jpeg
+AVATAR_PATH: str = os.environ.get('AVATAR_PATH')
+
+# (Optional). Only use this if you want a service did different from did:web
+SERVICE_DID: str = os.environ.get('SERVICE_DID')
+
+# (Optional). If your feed accepts interactions from clients
+ACCEPTS_INTERACTIONS: bool = _get_bool_env_var(os.environ.get('ACCEPTS_INTERACTIONS'))
+
+# (Optional). If your feed is a video feed
+IS_VIDEO_FEED: bool = _get_bool_env_var(os.environ.get('IS_VIDEO_FEED'))
+
+# -------------------------------------
+# NO NEED TO TOUCH ANYTHING BELOW HERE
+# -------------------------------------
+
+
+def main():
+    client = Client()
+    client.login(HANDLE, PASSWORD)
+
+    feed_did = SERVICE_DID
+    if not feed_did:
+        feed_did = f'did:web:{HOSTNAME}'
+
+    avatar_blob = None
+    if AVATAR_PATH:
+        with open(AVATAR_PATH, 'rb') as f:
+            avatar_data = f.read()
+            avatar_blob = client.upload_blob(avatar_data).blob
+
+    response = client.com.atproto.repo.put_record(models.ComAtprotoRepoPutRecord.Data(
+        repo=client.me.did,
+        collection=models.ids.AppBskyFeedGenerator,
+        rkey=RECORD_NAME,
+        record=models.AppBskyFeedGenerator.Record(
+            did=feed_did,
+            display_name=DISPLAY_NAME,
+            accepts_interactions=ACCEPTS_INTERACTIONS,
+            description=DESCRIPTION,
+            avatar=avatar_blob,
+            content_mode='app.bsky.feed.defs#contentModeVideo' if IS_VIDEO_FEED else None,
+            created_at=client.get_current_time_iso(),
+        )
+    ))
+
+    print('Successfully published!')
+    print('Feed URI (put in "FEED_URI" env var):', response.uri)
+
+
+if __name__ == '__main__':
+    main()

requirements.txt

···
+atproto==0.0.59
+peewee~=3.16.2
+Flask~=2.3.2
+python-dotenv~=1.0.0

server/__main__.py

···
+import logging
+
+from app import app
+from server.logger import logger
+
+if __name__ == '__main__':
+    # FOR DEBUG PURPOSE ONLY
+    logger.setLevel(logging.DEBUG)
+    app.run(host='127.0.0.1', port=8000, debug=True)

server/algos/__init__.py

···
+from . import feed
+
+algos = {
+    feed.uri: feed.handler
+}

server/algos/feed.py

···
+from datetime import datetime
+from typing import Optional
+
+from server import config
+from server.database import Post
+
+uri = config.FEED_URI
+CURSOR_EOF = 'eof'
+
+
+def handler(cursor: Optional[str], limit: int) -> dict:
+    posts = Post.select().order_by(Post.cid.desc()).order_by(Post.indexed_at.desc()).limit(limit)
+
+    if cursor:
+        if cursor == CURSOR_EOF:
+            return {
+                'cursor': CURSOR_EOF,
+                'feed': []
+            }
+        cursor_parts = cursor.split('::')
+        if len(cursor_parts) != 2:
+            raise ValueError('Malformed cursor')
+
+        indexed_at, cid = cursor_parts
+        indexed_at = datetime.fromtimestamp(int(indexed_at) / 1000)
+        posts = posts.where(((Post.indexed_at == indexed_at) & (Post.cid < cid)) | (Post.indexed_at < indexed_at))
+
+    feed = [{'post': post.uri} for post in posts]
+
+    cursor = CURSOR_EOF
+    last_post = posts[-1] if posts else None
+    if last_post:
+        cursor = f'{int(last_post.indexed_at.timestamp() * 1000)}::{last_post.cid}'
+
+    return {
+        'cursor': cursor,
+        'feed': feed
+    }

server/app.py

···
+import sys
+import signal
+import threading
+
+from server import config
+from server import data_stream
+
+from flask import Flask, jsonify, request
+
+from server.algos import algos
+from server.data_filter import operations_callback
+
+app = Flask(__name__)
+
+stream_stop_event = threading.Event()
+stream_thread = threading.Thread(
+    target=data_stream.run, args=(config.SERVICE_DID, operations_callback, stream_stop_event,)
+)
+stream_thread.start()
+
+
+def sigint_handler(*_):
+    print('Stopping data stream...')
+    stream_stop_event.set()
+    sys.exit(0)
+
+
+signal.signal(signal.SIGINT, sigint_handler)
+
+
[e-mail]('/')
+def index():
+    return 'ATProto Feed Generator powered by The AT Protocol SDK for Python (https://github.com/MarshalX/atproto).'
+
+
[e-mail]('/.well-known/did.json', methods=['GET'])
+def did_json():
+    if not config.SERVICE_DID.endswith(config.HOSTNAME):
+        return '', 404
+
+    return jsonify({
+        '@context': ['https://www.w3.org/ns/did/v1'],
+        'id': config.SERVICE_DID,
+        'service': [
+            {
+                'id': '#bsky_fg',
+                'type': 'BskyFeedGenerator',
+                'serviceEndpoint': f'https://{config.HOSTNAME}'
+            }
+        ]
+    })
+
+
[e-mail]('/xrpc/app.bsky.feed.describeFeedGenerator', methods=['GET'])
+def describe_feed_generator():
+    feeds = [{'uri': uri} for uri in algos.keys()]
+    response = {
+        'encoding': 'application/json',
+        'body': {
+            'did': config.SERVICE_DID,
+            'feeds': feeds
+        }
+    }
+    return jsonify(response)
+
+
[e-mail]('/xrpc/app.bsky.feed.getFeedSkeleton', methods=['GET'])
+def get_feed_skeleton():
+    feed = request.args.get('feed', default=None, type=str)
+    algo = algos.get(feed)
+    if not algo:
+        return 'Unsupported algorithm', 400
+
+    # Example of how to check auth if giving user-specific results:
+    """
+    from server.auth import AuthorizationError, validate_auth
+    try:
+        requester_did = validate_auth(request)
+    except AuthorizationError:
+        return 'Unauthorized', 401
+    """
+
+    try:
+        cursor = request.args.get('cursor', default=None, type=str)
+        limit = request.args.get('limit', default=20, type=int)
+        body = algo(cursor, limit)
+    except ValueError:
+        return 'Malformed cursor', 400
+
+    return jsonify(body)

server/auth.py

···
+from atproto import DidInMemoryCache, IdResolver, verify_jwt
+from atproto.exceptions import TokenInvalidSignatureError
+from flask import Request
+
+
+_CACHE = DidInMemoryCache()
+_ID_RESOLVER = IdResolver(cache=_CACHE)
+
+_AUTHORIZATION_HEADER_NAME = 'Authorization'
+_AUTHORIZATION_HEADER_VALUE_PREFIX = 'Bearer '
+
+
+class AuthorizationError(Exception):
+    ...
+
+
+def validate_auth(request: 'Request') -> str:
+    """Validate authorization header.
+
+    Args:
+        request: The request to validate.
+
+    Returns:
+        :obj:`str`: Requester DID.
+
+    Raises:
+        :obj:`AuthorizationError`: If the authorization header is invalid.
+    """
+    auth_header = request.headers.get(_AUTHORIZATION_HEADER_NAME)
+    if not auth_header:
+        raise AuthorizationError('Authorization header is missing')
+
+    if not auth_header.startswith(_AUTHORIZATION_HEADER_VALUE_PREFIX):
+        raise AuthorizationError('Invalid authorization header')
+
+    jwt = auth_header[len(_AUTHORIZATION_HEADER_VALUE_PREFIX) :].strip()
+
+    try:
+        return verify_jwt(jwt, _ID_RESOLVER.did.resolve_atproto_key).iss
+    except TokenInvalidSignatureError as e:
+        raise AuthorizationError('Invalid signature') from e

server/config.py

···
+import os
+import logging
+
+from dotenv import load_dotenv
+
+from server.logger import logger
+
+load_dotenv()
+
+SERVICE_DID = os.environ.get('SERVICE_DID')
+HOSTNAME = os.environ.get('HOSTNAME')
+FLASK_RUN_FROM_CLI = os.environ.get('FLASK_RUN_FROM_CLI')
+
+if FLASK_RUN_FROM_CLI:
+    logger.setLevel(logging.DEBUG)
+
+if not HOSTNAME:
+    raise RuntimeError('You should set "HOSTNAME" environment variable first.')
+
+if not SERVICE_DID:
+    SERVICE_DID = f'did:web:{HOSTNAME}'
+
+
+FEED_URI = os.environ.get('FEED_URI')
+if not FEED_URI:
+    raise RuntimeError('Publish your feed first (run publish_feed.py) to obtain Feed URI. '
+                       'Set this URI to "FEED_URI" environment variable.')
+
+
+def _get_bool_env_var(value: str) -> bool:
+    if value is None:
+        return False
+
+    normalized_value = value.strip().lower()
+    if normalized_value in {'1', 'true', 't', 'yes', 'y'}:
+        return True
+
+    return False
+
+
+IGNORE_ARCHIVED_POSTS = _get_bool_env_var(os.environ.get('IGNORE_ARCHIVED_POSTS'))
+IGNORE_REPLY_POSTS = _get_bool_env_var(os.environ.get('IGNORE_REPLY_POSTS'))

server/data_filter.py

···
+import datetime
+
+from collections import defaultdict
+
+from atproto import models
+
+from server import config
+from server.logger import logger
+from server.database import db, Post
+
+
+def is_archive_post(record: 'models.AppBskyFeedPost.Record') -> bool:
+    # Sometimes users will import old posts from Twitter/X which con flood a feed with
+    # old posts. Unfortunately, the only way to test for this is to look an old
+    # created_at date. However, there are other reasons why a post might have an old
+    # date, such as firehose or firehose consumer outages. It is up to you, the feed
+    # creator to weigh the pros and cons, amd and optionally include this function in
+    # your filter conditions, and adjust the threshold to your liking.
+    #
+    # See https://github.com/MarshalX/bluesky-feed-generator/pull/21
+
+    archived_threshold = datetime.timedelta(days=1)
+    created_at = datetime.datetime.fromisoformat(record.created_at)
+    now = datetime.datetime.now(datetime.UTC)
+
+    return now - created_at > archived_threshold
+
+
+def should_ignore_post(created_post: dict) -> bool:
+    record = created_post['record']
+    uri = created_post['uri']
+
+    if config.IGNORE_ARCHIVED_POSTS and is_archive_post(record):
+        logger.debug(f'Ignoring archived post: {uri}')
+        return True
+
+    if config.IGNORE_REPLY_POSTS and record.reply:
+        logger.debug(f'Ignoring reply post: {uri}')
+        return True
+
+    return False
+
+
+def operations_callback(ops: defaultdict) -> None:
+    # Here we can filter, process, run ML classification, etc.
+    # After our feed alg we can save posts into our DB
+    # Also, we should process deleted posts to remove them from our DB and keep it in sync
+
+    # for example, let's create our custom feed that will contain all posts that contains 'python' related text
+
+    posts_to_create = []
+    for created_post in ops[models.ids.AppBskyFeedPost]['created']:
+        author = created_post['author']
+        record = created_post['record']
+
+        post_with_images = isinstance(record.embed, models.AppBskyEmbedImages.Main)
+        post_with_video = isinstance(record.embed, models.AppBskyEmbedVideo.Main)
+        inlined_text = record.text.replace('\n', ' ')
+
+        # print all texts just as demo that data stream works
+        logger.debug(
+            f'NEW POST '
+            f'[CREATED_AT={record.created_at}]'
+            f'[AUTHOR={author}]'
+            f'[WITH_IMAGE={post_with_images}]'
+            f'[WITH_VIDEO={post_with_video}]'
+            f': {inlined_text}'
+        )
+
+        if should_ignore_post(created_post):
+            continue
+
+        # only python-related posts
+        if 'python' in record.text.lower():
+            reply_root = reply_parent = None
+            if record.reply:
+                reply_root = record.reply.root.uri
+                reply_parent = record.reply.parent.uri
+
+            post_dict = {
+                'uri': created_post['uri'],
+                'cid': created_post['cid'],
+                'reply_parent': reply_parent,
+                'reply_root': reply_root,
+            }
+            posts_to_create.append(post_dict)
+
+    posts_to_delete = ops[models.ids.AppBskyFeedPost]['deleted']
+    if posts_to_delete:
+        post_uris_to_delete = [post['uri'] for post in posts_to_delete]
+        Post.delete().where(Post.uri.in_(post_uris_to_delete))
+        logger.debug(f'Deleted from feed: {len(post_uris_to_delete)}')
+
+    if posts_to_create:
+        with db.atomic():
+            for post_dict in posts_to_create:
+                Post.create(**post_dict)
+        logger.debug(f'Added to feed: {len(posts_to_create)}')

server/data_stream.py

···
+import logging
+from collections import defaultdict
+
+from atproto import AtUri, CAR, firehose_models, FirehoseSubscribeReposClient, models, parse_subscribe_repos_message
+from atproto.exceptions import FirehoseError
+
+from server.database import SubscriptionState
+from server.logger import logger
+
+_INTERESTED_RECORDS = {
+    models.AppBskyFeedLike: models.ids.AppBskyFeedLike,
+    models.AppBskyFeedPost: models.ids.AppBskyFeedPost,
+    models.AppBskyGraphFollow: models.ids.AppBskyGraphFollow,
+}
+
+
+def _get_ops_by_type(commit: models.ComAtprotoSyncSubscribeRepos.Commit) -> defaultdict:
+    operation_by_type = defaultdict(lambda: {'created': [], 'deleted': []})
+
+    car = CAR.from_bytes(commit.blocks)
+    for op in commit.ops:
+        if op.action == 'update':
+            # we are not interested in updates
+            continue
+
+        uri = AtUri.from_str(f'at://{commit.repo}/{op.path}')
+
+        if op.action == 'create':
+            if not op.cid:
+                continue
+
+            create_info = {'uri': str(uri), 'cid': str(op.cid), 'author': commit.repo}
+
+            record_raw_data = car.blocks.get(op.cid)
+            if not record_raw_data:
+                continue
+
+            record = models.get_or_create(record_raw_data, strict=False)
+            if record is None:  # unknown record (out of bsky lexicon)
+                continue
+
+            for record_type, record_nsid in _INTERESTED_RECORDS.items():
+                if uri.collection == record_nsid and models.is_record_type(record, record_type):
+                    operation_by_type[record_nsid]['created'].append({'record': record, **create_info})
+                    break
+
+        if op.action == 'delete':
+            operation_by_type[uri.collection]['deleted'].append({'uri': str(uri)})
+
+    return operation_by_type
+
+
+def run(name, operations_callback, stream_stop_event=None):
+    while stream_stop_event is None or not stream_stop_event.is_set():
+        try:
+            _run(name, operations_callback, stream_stop_event)
+        except FirehoseError as e:
+            if logger.level == logging.DEBUG:
+                raise e
+            logger.error(f'Firehose error: {e}. Reconnecting to the firehose.')
+
+
+def _run(name, operations_callback, stream_stop_event=None):
+    state = SubscriptionState.get_or_none(SubscriptionState.service == name)
+
+    params = None
+    if state:
+        params = models.ComAtprotoSyncSubscribeRepos.Params(cursor=state.cursor)
+
+    client = FirehoseSubscribeReposClient(params)
+
+    if not state:
+        SubscriptionState.create(service=name, cursor=0)
+
+    def on_message_handler(message: firehose_models.MessageFrame) -> None:
+        # stop on next message if requested
+        if stream_stop_event and stream_stop_event.is_set():
+            client.stop()
+            return
+
+        commit = parse_subscribe_repos_message(message)
+        if not isinstance(commit, models.ComAtprotoSyncSubscribeRepos.Commit):
+            return
+
+        # update stored state every ~1k events
+        if commit.seq % 1000 == 0:  # lower value could lead to performance issues
+            logger.debug(f'Updated cursor for {name} to {commit.seq}')
+            client.update_params(models.ComAtprotoSyncSubscribeRepos.Params(cursor=commit.seq))
+            SubscriptionState.update(cursor=commit.seq).where(SubscriptionState.service == name).execute()
+
+        if not commit.blocks:
+            return
+
+        operations_callback(_get_ops_by_type(commit))
+
+    client.start(on_message_handler)

server/database.py

···
+from datetime import datetime
+
+import peewee
+
+db = peewee.SqliteDatabase('feed_database.db')
+
+
+class BaseModel(peewee.Model):
+    class Meta:
+        database = db
+
+
+class Post(BaseModel):
+    uri = peewee.CharField(index=True)
+    cid = peewee.CharField()
+    reply_parent = peewee.CharField(null=True, default=None)
+    reply_root = peewee.CharField(null=True, default=None)
+    indexed_at = peewee.DateTimeField(default=datetime.utcnow)
+
+
+class SubscriptionState(BaseModel):
+    service = peewee.CharField(unique=True)
+    cursor = peewee.BigIntegerField()
+
+
+if db.is_closed():
+    db.connect()
+    db.create_tables([Post, SubscriptionState])

server/logger.py

···
+import logging
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(level=logging.INFO)

setupvenv.sh

···
+#!/usr/bin/env bash
+
+set -e
+
+if [ ! -d "venv" ]; then
+  virtualenv venv || exit
+fi
+
+. venv/bin/activate
+
+pip install -r requirements.txt