···

       
53
       
 
       


     

       
54
       
 
       
    feed_manager = FeedManager()

     

       
55
       
 
       
    feed_manager.register(RapidFireFeed)

     

       
56
       
-
       
    feed_manager.register(PopularFeed)

     

       
57
       
 
       


     

       
58
       
 
       
    current_minute = None

     

       
59
       
 
       
    async for commit in firehose_events(firehose_manager):

     

···

       
53
       
 
       


     

       
54
       
 
       
    feed_manager = FeedManager()

     

       
55
       
 
       
    feed_manager.register(RapidFireFeed)

     

       
56
       
+
       
    # feed_manager.register(PopularFeed)

     

       
57
       
 
       


     

       
58
       
 
       
    current_minute = None

     

       
59
       
 
       
    async for commit in firehose_events(firehose_manager):

     

···

       
1
       
 
       
import os

     

       
2
       
 
       
import sys

     

       
3
       
-
       
import sqlite3

     

       
4
       
 
       


     

       
5
       
 
       
from . import BaseFeed

     

       
6
       
 
       


     
···

       
10
       
 
       
    FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/rapidfire'

     

       
11
       
 
       


     

       
12
       
 
       
    def __init__(self):

     

       
0
       
 
       

     

       
13
       
 
       
        if os.path.isdir('/dev/shm/'):

     

       
14
       
 
       
            os.makedirs('/dev/shm/feedgens/', exist_ok=True)

     

       
15
       
-
       
            self.db_cnx = sqlite3.connect('/dev/shm/feedgens/rapidfire.db')

     

       
16
       
 
       
        else:

     

       
17
       
-
       
            self.db_cnx = sqlite3.connect('db/rapidfire.db')

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
18
       
 
       


     

       
19
       
 
       
        with self.db_cnx:

     

       
20
       
-
       
            self.db_cnx.executescript(

     

       
21
       
-
       
                "pragma journal_mode = WAL;"

     

       
22
       
-
       
                "pragma synchronous = OFF;"

     

       
23
       
-
       
                "pragma wal_autocheckpoint = 0;"

     

       
24
       
-
       
                "create table if not exists posts (uri text, create_ts timestamp, lang text);"

     

       
25
       
-
       
                "create index if not exists create_ts_idx on posts(create_ts);"

     

       
26
       
-
       
            )

     

       
27
       
-
       


     

       
28
       
-
       
        self.checkpoint = 0

     

       
29
       
 
       


     

       
30
       
 
       
    def process_commit(self, commit):

     

       
31
       
 
       
        op = commit['op']

     
···

       
48
       
 
       
            path = op['path']

     

       
49
       
 
       
            post_uri = f'at://{repo}/{path}'

     

       
50
       
 
       
            ts = record['createdAt']

     

       
51
       
-
       
            self.checkpoint += 1

     

       
52
       
 
       


     

       
53
       
 
       
            with self.db_cnx:

     

       
54
       
 
       
                langs = record.get('langs') or ['']

     
···

       
58
       
 
       
                        dict(uri=post_uri, ts=ts, lang=lang)

     

       
59
       
 
       
                    )

     

       
60
       
 
       


     

       
61
       
-
       
                self.db_cnx.execute(

     

       
62
       
-
       
                    "delete from posts where strftime('%s', create_ts) < strftime('%s', 'now', '-15 minutes')"

     

       
63
       
-
       
                )

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
64
       
 
       


     

       
65
       
-
       
            if self.checkpoint % 100 == 0:

     

       
66
       
-
       
                sys.stdout.write('rapidfire: checkpoint\n')

     

       
67
       
-
       
                sys.stdout.flush()

     

       
68
       
-
       
                self.db_cnx.execute("pragma wal_checkpoint(TRUNCATE)")

     

       
69
       
 
       


     

       
70
       
 
       
    def serve_feed(self, limit, offset, langs):

     

       
71
       
 
       
        if '*' in langs:

     

···

       
1
       
 
       
import os

     

       
2
       
 
       
import sys

     

       
3
       
+
       
import apsw

     

       
4
       
 
       


     

       
5
       
 
       
from . import BaseFeed

     

       
6
       
 
       


     
···

       
10
       
 
       
    FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/rapidfire'

     

       
11
       
 
       


     

       
12
       
 
       
    def __init__(self):

     

       
13
       
+
       
        db_fname = ''

     

       
14
       
 
       
        if os.path.isdir('/dev/shm/'):

     

       
15
       
 
       
            os.makedirs('/dev/shm/feedgens/', exist_ok=True)

     

       
16
       
+
       
            db_fname = '/dev/shm/feedgens/rapidfire.db'

     

       
17
       
 
       
        else:

     

       
18
       
+
       
            db_fname = 'db/rapidfire.db'

     

       
19
       
+
       


     

       
20
       
+
       
        self.db_cnx = apsw.Connection(db_fname)

     

       
21
       
+
       
        self.db_cnx.pragma('journal_mode', 'WAL')

     

       
22
       
+
       
        self.db_cnx.pragma('synchronous', 'OFF')

     

       
23
       
+
       
        self.db_cnx.pragma('wal_autocheckpoint', '0')

     

       
24
       
 
       


     

       
25
       
 
       
        with self.db_cnx:

     

       
26
       
+
       
            self.db_cnx.execute("""

     

       
27
       
+
       
            create table if not exists posts (uri text, create_ts timestamp, lang text);

     

       
28
       
+
       
            create index if not exists create_ts_idx on posts(create_ts);

     

       
29
       
+
       
            """)

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
30
       
 
       


     

       
31
       
 
       
    def process_commit(self, commit):

     

       
32
       
 
       
        op = commit['op']

     
···

       
49
       
 
       
            path = op['path']

     

       
50
       
 
       
            post_uri = f'at://{repo}/{path}'

     

       
51
       
 
       
            ts = record['createdAt']

     

       
0
       
 
       

     

       
52
       
 
       


     

       
53
       
 
       
            with self.db_cnx:

     

       
54
       
 
       
                langs = record.get('langs') or ['']

     
···

       
58
       
 
       
                        dict(uri=post_uri, ts=ts, lang=lang)

     

       
59
       
 
       
                    )

     

       
60
       
 
       


     

       
61
       
+
       
    def run_tasks_minute(self):

     

       
62
       
+
       
        sys.stdout.write('rapidfire: running minute tasks\n')

     

       
63
       
+
       
        sys.stdout.flush()

     

       
64
       
+
       


     

       
65
       
+
       
        with self.db_cnx:

     

       
66
       
+
       
            self.db_cnx.execute(

     

       
67
       
+
       
                "delete from posts where strftime('%s', create_ts) < strftime('%s', 'now', '-15 minutes')"

     

       
68
       
+
       
            )

     

       
69
       
 
       


     

       
70
       
+
       
        self.db_cnx.pragma('wal_checkpoint(TRUNCATE)')

     

       
0
       
 
       

     

       
0
       
 
       

     

       
0
       
 
       

     

       
71
       
 
       


     

       
72
       
 
       
    def serve_feed(self, limit, offset, langs):

     

       
73
       
 
       
        if '*' in langs:

     

···

       
11
       
 
       
def get_feed_skeleton():

     

       
12
       
 
       
    manager = FeedManager()

     

       
13
       
 
       
    manager.register(RapidFireFeed)

     

       
14
       
-
       
    manager.register(PopularFeed)

     

       
15
       
 
       


     

       
16
       
 
       
    try:

     

       
17
       
 
       
        limit = int(request.args.get('limit', 50))

     

···

       
11
       
 
       
def get_feed_skeleton():

     

       
12
       
 
       
    manager = FeedManager()

     

       
13
       
 
       
    manager.register(RapidFireFeed)

     

       
14
       
+
       
    # manager.register(PopularFeed)

     

       
15
       
 
       


     

       
16
       
 
       
    try:

     

       
17
       
 
       
        limit = int(request.args.get('limit', 50))