edavis.dev
-47
feeds/mostliked.py
-47
feeds/mostliked.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
6
-
from . import BaseFeed
7
-
8
-
class MostLikedFeed(BaseFeed):
9
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/most-liked'
10
-
11
-
def __init__(self):
12
-
self.db_cnx = apsw.Connection('db/mostliked.db')
13
-
self.db_cnx.pragma('foreign_keys', True)
14
-
self.db_cnx.pragma('journal_mode', 'WAL')
15
-
16
-
def generate_sql(self, limit, offset, langs):
17
-
bindings = []
18
-
sql = """
19
-
select posts.uri, create_ts, create_ts - unixepoch('now', '-24 hours') as ttl, likes, lang
20
-
from posts
21
-
left join langs on posts.uri = langs.uri
22
-
where
23
-
"""
24
-
if not langs:
25
-
sql += " 1=1 "
26
-
else:
27
-
lang_values = list(langs.values())
28
-
bindings.extend(lang_values)
29
-
sql += " OR ".join(['lang = ?'] * len(lang_values))
30
-
sql += """
31
-
order by likes desc, create_ts desc
32
-
limit ? offset ?
33
-
"""
34
-
bindings.extend([limit, offset])
35
-
return sql, bindings
36
-
37
-
def serve_feed(self, limit, offset, langs):
38
-
sql, bindings = self.generate_sql(limit, offset, langs)
39
-
cur = self.db_cnx.execute(sql, bindings)
40
-
return [row[0] for row in cur]
41
-
42
-
def serve_feed_debug(self, limit, offset, langs):
43
-
sql, bindings = self.generate_sql(limit, offset, langs)
44
-
return apsw.ext.format_query_table(
45
-
self.db_cnx, sql, bindings,
46
-
string_sanitize=2, text_width=9999, use_unicode=True
47
-
)
-16
service/feedgen.service
-16
service/feedgen.service
···
1
-
[Unit]
2
-
Description=Bsky Feedgen
3
-
After=network.target syslog.target
4
-
5
-
[Service]
6
-
Type=simple
7
-
User=eric
8
-
WorkingDirectory=/home/eric/bsky-tools
9
-
ExecStart=/home/eric/.local/bin/pipenv run ./feedgen.py
10
-
TimeoutSec=15
11
-
Restart=on-failure
12
-
RestartSec=60
13
-
StandardOutput=journal
14
-
15
-
[Install]
16
-
WantedBy=multi-user.target
-16
service/feedweb.service
-16
service/feedweb.service
···
1
-
[Unit]
2
-
Description=Bsky Feedweb
3
-
After=network.target syslog.target
4
-
5
-
[Service]
6
-
Type=simple
7
-
User=eric
8
-
WorkingDirectory=/home/eric/bsky-tools
9
-
ExecStart=/home/eric/.local/bin/pipenv run gunicorn -w 1 -b 127.0.0.1:9060 feedweb:app
10
-
TimeoutSec=15
11
-
Restart=on-failure
12
-
RestartSec=1
13
-
StandardOutput=journal
14
-
15
-
[Install]
16
-
WantedBy=multi-user.target
+3
cmd/bsky-users/schema.sql
+3
cmd/bsky-users/schema.sql
-76
feedgen.py
-76
feedgen.py
···
1
-
#!/usr/bin/env python3
2
-
3
-
import asyncio
4
-
from io import BytesIO
5
-
import json
6
-
import logging
7
-
import signal
8
-
9
-
from atproto import CAR
10
-
import dag_cbor
11
-
import websockets
12
-
13
-
from feed_manager import feed_manager
14
-
15
-
logging.basicConfig(
16
-
format='%(asctime)s - %(levelname)-5s - %(name)-20s - %(message)s',
17
-
level=logging.DEBUG
18
-
)
19
-
logging.getLogger('').setLevel(logging.WARNING)
20
-
logging.getLogger('feeds').setLevel(logging.DEBUG)
21
-
logging.getLogger('firehose').setLevel(logging.DEBUG)
22
-
logging.getLogger('feedgen').setLevel(logging.DEBUG)
23
-
24
-
logger = logging.getLogger('feedgen')
25
-
26
-
async def firehose_events():
27
-
relay_url = 'ws://localhost:6008/subscribe'
28
-
29
-
logger = logging.getLogger('feeds.events')
30
-
logger.info(f'opening websocket connection to {relay_url}')
31
-
32
-
async with websockets.connect(relay_url, ping_timeout=60) as firehose:
33
-
while True:
34
-
payload = BytesIO(await firehose.recv())
35
-
yield json.load(payload)
36
-
37
-
async def main():
38
-
event_count = 0
39
-
40
-
async for commit in firehose_events():
41
-
feed_manager.process_commit(commit)
42
-
event_count += 1
43
-
if event_count % 2500 == 0:
44
-
feed_manager.commit_changes()
45
-
46
-
def handle_exception(loop, context):
47
-
msg = context.get("exception", context["message"])
48
-
logger.error(f"Caught exception: {msg}")
49
-
logger.info("Shutting down...")
50
-
asyncio.create_task(shutdown(loop))
51
-
52
-
async def shutdown(loop, signal=None):
53
-
if signal:
54
-
logger.info(f'received exit signal {signal.name}')
55
-
feed_manager.stop_all()
56
-
tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
57
-
[task.cancel() for task in tasks]
58
-
logger.info(f'cancelling {len(tasks)} outstanding tasks')
59
-
await asyncio.gather(*tasks, return_exceptions=True)
60
-
loop.stop()
61
-
62
-
if __name__ == '__main__':
63
-
loop = asyncio.get_event_loop()
64
-
catch_signals = (signal.SIGTERM, signal.SIGINT)
65
-
for sig in catch_signals:
66
-
loop.add_signal_handler(
67
-
sig,
68
-
lambda s=sig: asyncio.create_task(shutdown(loop, signal=s))
69
-
)
70
-
loop.set_exception_handler(handle_exception)
71
-
72
-
try:
73
-
loop.create_task(main())
74
-
loop.run_forever()
75
-
finally:
76
-
loop.close()
-65
feeds/__init__.py
-65
feeds/__init__.py
···
1
-
from datetime import datetime, timezone, timedelta
2
-
3
-
class BaseFeed:
4
-
def process_commit(self, commit):
5
-
raise NotImplementedError
6
-
7
-
def serve_feed(self, limit, offset, langs):
8
-
raise NotImplementedError
9
-
10
-
def serve_wildcard_feed(self, feed_uri, limit, offset, langs):
11
-
raise NotImplementedError
12
-
13
-
def commit_changes(self):
14
-
raise NotImplementedError
15
-
16
-
def parse_timestamp(self, timestamp):
17
-
# https://atproto.com/specs/lexicon#datetime
18
-
formats = {
19
-
# preferred
20
-
'1985-04-12T23:20:50.123Z': '%Y-%m-%dT%H:%M:%S.%f%z',
21
-
# '1985-04-12T23:20:50.123456Z': '%Y-%m-%dT%H:%M:%S.%f%z',
22
-
# '1985-04-12T23:20:50.120Z': '%Y-%m-%dT%H:%M:%S.%f%z',
23
-
# '1985-04-12T23:20:50.120000Z': '%Y-%m-%dT%H:%M:%S.%f%z',
24
-
25
-
# supported
26
-
# '1985-04-12T23:20:50.12345678912345Z': '',
27
-
'1985-04-12T23:20:50Z': '%Y-%m-%dT%H:%M:%S%z',
28
-
# '1985-04-12T23:20:50.0Z': '%Y-%m-%dT%H:%M:%S.%f%z',
29
-
# '1985-04-12T23:20:50.123+00:00': '%Y-%m-%dT%H:%M:%S.%f%z',
30
-
# '1985-04-12T23:20:50.123-07:00': '%Y-%m-%dT%H:%M:%S.%f%z',
31
-
}
32
-
33
-
for format in formats.values():
34
-
try:
35
-
ts = datetime.strptime(timestamp, format)
36
-
except ValueError:
37
-
continue
38
-
else:
39
-
return ts
40
-
41
-
return datetime.now(timezone.utc)
42
-
43
-
def safe_timestamp(self, timestamp):
44
-
utc_now = datetime.now(timezone.utc)
45
-
if timestamp is None:
46
-
return utc_now
47
-
48
-
parsed = self.parse_timestamp(timestamp)
49
-
if parsed.timestamp() <= 0:
50
-
return utc_now
51
-
elif parsed - timedelta(minutes=2) < utc_now:
52
-
return parsed
53
-
elif parsed > utc_now:
54
-
return utc_now
55
-
56
-
def transaction_begin(self, db):
57
-
if not db.in_transaction:
58
-
db.execute('BEGIN')
59
-
60
-
def transaction_commit(self, db):
61
-
if db.in_transaction:
62
-
db.execute('COMMIT')
63
-
64
-
def wal_checkpoint(self, db, mode='PASSIVE'):
65
-
return db.execute(f'PRAGMA wal_checkpoint({mode})')
-100
feeds/battle.py
-100
feeds/battle.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
import grapheme
6
-
7
-
from . import BaseFeed
8
-
9
-
class BattleFeed(BaseFeed):
10
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/battle'
11
-
12
-
def __init__(self):
13
-
self.db_cnx = apsw.Connection('db/battle.db')
14
-
self.db_cnx.pragma('journal_mode', 'WAL')
15
-
self.db_cnx.pragma('wal_autocheckpoint', '0')
16
-
17
-
with self.db_cnx:
18
-
self.db_cnx.execute("""
19
-
create table if not exists posts (
20
-
uri text,
21
-
grapheme_length integer,
22
-
create_ts timestamp,
23
-
lang text
24
-
);
25
-
create unique index if not exists ll_idx on posts(grapheme_length, lang);
26
-
""")
27
-
28
-
self.logger = logging.getLogger('feeds.battle')
29
-
30
-
def process_commit(self, commit):
31
-
if commit['opType'] != 'c':
32
-
return
33
-
34
-
if commit['collection'] != 'app.bsky.feed.post':
35
-
return
36
-
37
-
record = commit.get('record')
38
-
if record is None:
39
-
return
40
-
41
-
repo = commit['did']
42
-
rkey = commit['rkey']
43
-
post_uri = f'at://{repo}/app.bsky.feed.post/{rkey}'
44
-
length = grapheme.length(record.get('text', ''))
45
-
ts = self.safe_timestamp(record.get('createdAt')).timestamp()
46
-
47
-
self.transaction_begin(self.db_cnx)
48
-
49
-
langs = record.get('langs') or ['']
50
-
for lang in langs:
51
-
self.db_cnx.execute("""
52
-
insert into posts(uri, grapheme_length, create_ts, lang)
53
-
values(:uri, :length, :ts, :lang)
54
-
on conflict do update set uri = :uri, create_ts = :ts
55
-
""", dict(uri=post_uri, length=length, ts=ts, lang=lang))
56
-
57
-
def commit_changes(self):
58
-
self.logger.debug('committing changes')
59
-
self.transaction_commit(self.db_cnx)
60
-
self.wal_checkpoint(self.db_cnx, 'RESTART')
61
-
62
-
def serve_feed(self, limit, offset, langs):
63
-
if '*' in langs:
64
-
cur = self.db_cnx.execute("""
65
-
select uri
66
-
from posts
67
-
order by grapheme_length asc
68
-
limit :limit offset :offset
69
-
""", dict(limit=limit, offset=offset))
70
-
return [uri for (uri,) in cur]
71
-
else:
72
-
lang_values = list(langs.values())
73
-
lang_selects = ['select uri, grapheme_length from posts where lang = ?'] * len(lang_values)
74
-
lang_clause = ' union '.join(lang_selects)
75
-
cur = self.db_cnx.execute(
76
-
lang_clause + ' order by grapheme_length asc limit ? offset ?',
77
-
[*lang_values, limit, offset]
78
-
)
79
-
return [uri for (uri, grapheme_length) in cur]
80
-
81
-
def serve_feed_debug(self, limit, offset, langs):
82
-
if '*' in langs:
83
-
query = """
84
-
select *, unixepoch('now') - create_ts as age_seconds
85
-
from posts
86
-
order by grapheme_length asc
87
-
limit :limit offset :offset
88
-
"""
89
-
bindings = [limit, offset]
90
-
else:
91
-
lang_values = list(langs.values())
92
-
lang_selects = ["select *, unixepoch('now') - create_ts as age_seconds from posts where lang = ?"] * len(lang_values)
93
-
lang_clause = ' union '.join(lang_selects)
94
-
query = lang_clause + ' order by grapheme_length asc limit ? offset ?'
95
-
bindings = [*lang_values, limit, offset]
96
-
97
-
return apsw.ext.format_query_table(
98
-
self.db_cnx, query, bindings,
99
-
string_sanitize=2, text_width=9999, use_unicode=True
100
-
)
-107
feeds/homeruns.py
-107
feeds/homeruns.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
6
-
from . import BaseFeed
7
-
8
-
MLBHRS_DID = 'did:plc:pnksqegntq5t3o7pusp2idx3'
9
-
10
-
TEAM_ABBR_LOOKUP = {
11
-
"OAK":"OaklandAthletics",
12
-
"PIT":"PittsburghPirates",
13
-
"SDN":"SanDiegoPadres",
14
-
"SEA":"SeattleMariners",
15
-
"SFN":"SanFranciscoGiants",
16
-
"SLN":"StLouisCardinals",
17
-
"TBA":"TampaBayRays",
18
-
"TEX":"TexasRangers",
19
-
"TOR":"TorontoBlueJays",
20
-
"MIN":"MinnesotaTwins",
21
-
"PHI":"PhiladelphiaPhillies",
22
-
"ATL":"AtlantaBraves",
23
-
"CHA":"ChicagoWhiteSox",
24
-
"MIA":"MiamiMarlins",
25
-
"NYA":"NewYorkYankees",
26
-
"MIL":"MilwaukeeBrewers",
27
-
"LAA":"LosAngelesAngels",
28
-
"ARI":"ArizonaDiamondbacks",
29
-
"BAL":"BaltimoreOrioles",
30
-
"BOS":"BostonRedSox",
31
-
"CHN":"ChicagoCubs",
32
-
"CIN":"CincinnatiReds",
33
-
"CLE":"ClevelandGuardians",
34
-
"COL":"ColoradoRockies",
35
-
"DET":"DetroitTigers",
36
-
"HOU":"HoustonAstros",
37
-
"KCA":"KansasCityRoyals",
38
-
"LAN":"LosAngelesDodgers",
39
-
"WAS":"WashingtonNationals",
40
-
"NYN":"NewYorkMets",
41
-
}
42
-
43
-
class HomeRunsTeamFeed(BaseFeed):
44
-
FEED_URI = 'at://did:plc:pnksqegntq5t3o7pusp2idx3/app.bsky.feed.generator/team:*'
45
-
46
-
def __init__(self):
47
-
self.db_cnx = apsw.Connection('db/homeruns.db')
48
-
self.db_cnx.pragma('journal_mode', 'WAL')
49
-
self.db_cnx.pragma('wal_autocheckpoint', '0')
50
-
51
-
with self.db_cnx:
52
-
self.db_cnx.execute("""
53
-
create table if not exists posts (uri text, tag text);
54
-
create index if not exists tag_idx on posts(tag);
55
-
""")
56
-
57
-
self.logger = logging.getLogger('feeds.homeruns')
58
-
59
-
def process_commit(self, commit):
60
-
if commit['did'] != MLBHRS_DID:
61
-
return
62
-
63
-
if commit['opType'] != 'c':
64
-
return
65
-
66
-
if commit['collection'] != 'app.bsky.feed.post':
67
-
return
68
-
69
-
record = commit.get('record')
70
-
if record is None:
71
-
return
72
-
73
-
uri = 'at://{repo}/app.bsky.feed.post/{rkey}'.format(
74
-
repo = commit['did'],
75
-
rkey = commit['rkey']
76
-
)
77
-
tags = record.get('tags', [])
78
-
79
-
self.logger.debug(f'adding {uri!r} under {tags!r}')
80
-
81
-
with self.db_cnx:
82
-
for tag in tags:
83
-
self.db_cnx.execute(
84
-
"insert into posts (uri, tag) values (:uri, :tag)",
85
-
dict(uri=uri, tag=tag)
86
-
)
87
-
88
-
def commit_changes(self):
89
-
self.logger.debug('committing changes')
90
-
self.wal_checkpoint(self.db_cnx, 'RESTART')
91
-
92
-
def serve_wildcard_feed(self, feed_uri, limit, offset, langs):
93
-
prefix, sep, team_abbr = feed_uri.rpartition(':')
94
-
team_tag = TEAM_ABBR_LOOKUP[team_abbr]
95
-
96
-
cur = self.db_cnx.execute("""
97
-
select uri
98
-
from posts
99
-
where tag = :tag
100
-
order by uri desc
101
-
limit :limit offset :offset
102
-
""", dict(tag=team_tag, limit=limit, offset=offset))
103
-
104
-
return [uri for (uri,) in cur]
105
-
106
-
def serve_wildcard_feed_debug(self, feed_uri, limit, offset, langs):
107
-
pass
-90
feeds/popqp.py
-90
feeds/popqp.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
6
-
from . import BaseFeed
7
-
8
-
class PopularQuotePostsFeed(BaseFeed):
9
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/popqp'
10
-
SERVE_FEED_QUERY = """
11
-
select uri, create_ts, update_ts, quote_count, exp( -1 * ( ( unixepoch('now') - create_ts ) / 10800.0 ) ) as decay,
12
-
quote_count * exp( -1 * ( ( unixepoch('now') - create_ts ) / 10800.0 ) ) as score
13
-
from posts
14
-
order by quote_count * exp( -1 * ( ( unixepoch('now') - create_ts ) / 10800.0 ) ) desc
15
-
limit :limit offset :offset
16
-
"""
17
-
DELETE_OLD_POSTS_QUERY = """
18
-
delete from posts where
19
-
quote_count * exp( -1 * ( ( unixepoch('now') - create_ts ) / 10800.0 ) ) < 1.0
20
-
and create_ts < unixepoch('now', '-24 hours')
21
-
"""
22
-
23
-
def __init__(self):
24
-
self.db_cnx = apsw.Connection('db/popqp.db')
25
-
self.db_cnx.pragma('journal_mode', 'WAL')
26
-
self.db_cnx.pragma('wal_autocheckpoint', '0')
27
-
28
-
with self.db_cnx:
29
-
self.db_cnx.execute("""
30
-
create table if not exists posts (
31
-
uri text, create_ts timestamp, update_ts timestamp, quote_count int
32
-
);
33
-
create unique index if not exists uri_idx on posts(uri);
34
-
""")
35
-
36
-
self.logger = logging.getLogger('feeds.popqp')
37
-
38
-
def process_commit(self, commit):
39
-
if commit['opType'] != 'c':
40
-
return
41
-
42
-
if commit['collection'] != 'app.bsky.feed.post':
43
-
return
44
-
45
-
record = commit.get('record')
46
-
if record is None:
47
-
return
48
-
49
-
embed = record.get('embed')
50
-
if embed is None:
51
-
return
52
-
53
-
embed_type = embed.get('$type')
54
-
if embed_type == 'app.bsky.embed.record':
55
-
quote_post_uri = embed['record']['uri']
56
-
elif embed_type == 'app.bsky.embed.recordWithMedia':
57
-
quote_post_uri = embed['record']['record']['uri']
58
-
else:
59
-
return
60
-
61
-
ts = self.safe_timestamp(record.get('createdAt')).timestamp()
62
-
self.transaction_begin(self.db_cnx)
63
-
64
-
self.db_cnx.execute("""
65
-
insert into posts (uri, create_ts, update_ts, quote_count)
66
-
values (:uri, :ts, :ts, 1)
67
-
on conflict (uri) do
68
-
update set quote_count = quote_count + 1, update_ts = :ts
69
-
""", dict(uri=quote_post_uri, ts=ts))
70
-
71
-
def delete_old_posts(self):
72
-
self.db_cnx.execute(self.DELETE_OLD_POSTS_QUERY)
73
-
self.logger.debug('deleted {} old posts'.format(self.db_cnx.changes()))
74
-
75
-
def commit_changes(self):
76
-
self.delete_old_posts()
77
-
self.logger.debug('committing changes')
78
-
self.transaction_commit(self.db_cnx)
79
-
self.wal_checkpoint(self.db_cnx, 'RESTART')
80
-
81
-
def serve_feed(self, limit, offset, langs):
82
-
cur = self.db_cnx.execute(self.SERVE_FEED_QUERY, dict(limit=limit, offset=offset))
83
-
return [row[0] for row in cur]
84
-
85
-
def serve_feed_debug(self, limit, offset, langs):
86
-
bindings = dict(limit=limit, offset=offset)
87
-
return apsw.ext.format_query_table(
88
-
self.db_cnx, self.SERVE_FEED_QUERY, bindings,
89
-
string_sanitize=2, text_width=9999, use_unicode=True
90
-
)
-98
feeds/rapidfire.py
-98
feeds/rapidfire.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
import grapheme
6
-
7
-
from . import BaseFeed
8
-
9
-
MAX_TEXT_LENGTH = 140
10
-
11
-
class RapidFireFeed(BaseFeed):
12
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/rapidfire'
13
-
14
-
def __init__(self):
15
-
self.db_cnx = apsw.Connection('db/rapidfire.db')
16
-
self.db_cnx.pragma('journal_mode', 'WAL')
17
-
self.db_cnx.pragma('wal_autocheckpoint', '0')
18
-
19
-
with self.db_cnx:
20
-
self.db_cnx.execute("""
21
-
create table if not exists posts (uri text, create_ts timestamp, lang text);
22
-
create index if not exists create_ts_idx on posts(create_ts);
23
-
""")
24
-
25
-
self.logger = logging.getLogger('feeds.rapidfire')
26
-
27
-
def process_commit(self, commit):
28
-
if commit['opType'] != 'c':
29
-
return
30
-
31
-
if commit['collection'] != 'app.bsky.feed.post':
32
-
return
33
-
34
-
record = commit.get('record')
35
-
if record is None:
36
-
return
37
-
38
-
if all([
39
-
grapheme.length(record.get('text', '')) <= MAX_TEXT_LENGTH,
40
-
record.get('reply') is None,
41
-
record.get('embed') is None,
42
-
record.get('facets') is None
43
-
]):
44
-
repo = commit['did']
45
-
rkey = commit['rkey']
46
-
post_uri = f'at://{repo}/app.bsky.feed.post/{rkey}'
47
-
ts = self.safe_timestamp(record.get('createdAt')).timestamp()
48
-
49
-
self.transaction_begin(self.db_cnx)
50
-
51
-
langs = record.get('langs') or ['']
52
-
for lang in langs:
53
-
self.db_cnx.execute(
54
-
'insert into posts (uri, create_ts, lang) values (:uri, :ts, :lang)',
55
-
dict(uri=post_uri, ts=ts, lang=lang)
56
-
)
57
-
58
-
def delete_old_posts(self):
59
-
self.db_cnx.execute(
60
-
"delete from posts where create_ts < unixepoch('now', '-15 minutes')"
61
-
)
62
-
self.logger.debug('deleted {} old posts'.format(self.db_cnx.changes()))
63
-
64
-
def commit_changes(self):
65
-
self.delete_old_posts()
66
-
self.logger.debug('committing changes')
67
-
self.transaction_commit(self.db_cnx)
68
-
self.wal_checkpoint(self.db_cnx, 'RESTART')
69
-
70
-
def serve_feed(self, limit, offset, langs):
71
-
if '*' in langs:
72
-
cur = self.db_cnx.execute(
73
-
"select uri from posts order by create_ts desc limit :limit offset :offset",
74
-
dict(limit=limit, offset=offset)
75
-
)
76
-
return [uri for (uri,) in cur]
77
-
else:
78
-
lang_values = list(langs.values())
79
-
lang_selects = ['select uri, create_ts from posts where lang = ?'] * len(lang_values)
80
-
lang_clause = ' union '.join(lang_selects)
81
-
cur = self.db_cnx.execute(
82
-
lang_clause + ' order by create_ts desc limit ? offset ?',
83
-
[*lang_values, limit, offset]
84
-
)
85
-
return [uri for (uri, create_ts) in cur]
86
-
87
-
def serve_feed_debug(self, limit, offset, langs):
88
-
query = """
89
-
select *, unixepoch('now') - create_ts as age_seconds
90
-
from posts
91
-
order by create_ts desc
92
-
limit :limit offset :offset
93
-
"""
94
-
bindings = dict(limit=limit, offset=offset)
95
-
return apsw.ext.format_query_table(
96
-
self.db_cnx, query, bindings,
97
-
string_sanitize=2, text_width=9999, use_unicode=True
98
-
)
-144
feeds/ratio.py
-144
feeds/ratio.py
···
1
-
import logging
2
-
3
-
import apsw
4
-
import apsw.ext
5
-
6
-
from . import BaseFeed
7
-
8
-
class RatioFeed(BaseFeed):
9
-
FEED_URI = 'at://did:plc:4nsduwlpivpuur4mqkbfvm6a/app.bsky.feed.generator/ratio'
10
-
SERVE_FEED_QUERY = """
11
-
with served as (
12
-
select
13
-
uri,
14
-
create_ts,
15
-
( unixepoch('now') - create_ts ) as age_seconds,
16
-
replies,
17
-
quoteposts,
18
-
likes,
19
-
reposts,
20
-
( replies + quoteposts ) / ( likes + reposts + 1 ) as ratio,
21
-
exp( -1 * ( ( unixepoch('now') - create_ts ) / ( 3600.0 * 16 ) ) ) as decay
22
-
from posts
23
-
)
24
-
select
25
-
*,
26
-
( ratio * decay ) as score
27
-
from served
28
-
where replies > 15 and ratio > 2.5
29
-
order by score desc
30
-
limit :limit offset :offset
31
-
"""
32
-
DELETE_OLD_POSTS_QUERY = """
33
-
delete from posts
34
-
where
35
-
create_ts < unixepoch('now', '-5 days')
36
-
"""
37
-
38
-
def __init__(self):
39
-
self.db_cnx = apsw.Connection('db/ratio.db')
40
-
self.db_cnx.pragma('journal_mode', 'WAL')
41
-
self.db_cnx.pragma('wal_autocheckpoint', '0')
42
-
43
-
with self.db_cnx:
44
-
self.db_cnx.execute("""
45
-
create table if not exists posts (
46
-
uri text, create_ts timestamp,
47
-
replies float, likes float, reposts float, quoteposts float
48
-
);
49
-
create unique index if not exists uri_idx on posts(uri);
50
-
""")
51
-
52
-
self.logger = logging.getLogger('feeds.ratio')
53
-
54
-
def process_commit(self, commit):
55
-
if commit['opType'] != 'c':
56
-
return
57
-
58
-
subject_uri = None
59
-
is_reply = False
60
-
is_quotepost = False
61
-
62
-
if commit['collection'] in {'app.bsky.feed.like', 'app.bsky.feed.repost'}:
63
-
record = commit.get('record')
64
-
ts = self.safe_timestamp(record.get('createdAt')).timestamp()
65
-
try:
66
-
subject_uri = record['subject']['uri']
67
-
except KeyError:
68
-
return
69
-
elif commit['collection'] == 'app.bsky.feed.post':
70
-
record = commit.get('record')
71
-
ts = self.safe_timestamp(record.get('createdAt')).timestamp()
72
-
if record.get('reply') is not None:
73
-
is_reply = True
74
-
try:
75
-
subject_uri = record['reply']['parent']['uri']
76
-
except KeyError:
77
-
return
78
-
79
-
# only count non-OP replies
80
-
if subject_uri.startswith('at://' + commit['did']):
81
-
return
82
-
83
-
elif record.get('embed') is not None:
84
-
is_quotepost = True
85
-
t = record['embed']['$type']
86
-
if t == 'app.bsky.embed.record':
87
-
try:
88
-
subject_uri = record['embed']['record']['uri']
89
-
except KeyError:
90
-
return
91
-
elif t == 'app.bsky.embed.recordWithMedia':
92
-
try:
93
-
subject_uri = record['embed']['record']['record']['uri']
94
-
except KeyError:
95
-
return
96
-
97
-
if subject_uri is None:
98
-
return
99
-
100
-
params = {
101
-
'uri': subject_uri,
102
-
'ts': ts,
103
-
'is_reply': int(is_reply),
104
-
'is_like': int(commit['collection'] == 'app.bsky.feed.like'),
105
-
'is_repost': int(commit['collection'] == 'app.bsky.feed.repost'),
106
-
'is_quotepost': int(is_quotepost),
107
-
}
108
-
109
-
self.transaction_begin(self.db_cnx)
110
-
111
-
self.db_cnx.execute("""
112
-
insert into posts(uri, create_ts, replies, likes, reposts, quoteposts)
113
-
values (:uri, :ts,
114
-
case when :is_reply then 1 else 0 end,
115
-
case when :is_like then 1 else 0 end,
116
-
case when :is_repost then 1 else 0 end,
117
-
case when :is_quotepost then 1 else 0 end)
118
-
on conflict(uri)
119
-
do update set
120
-
replies = replies + case when :is_reply then 1 else 0 end,
121
-
likes = likes + case when :is_like then 1 else 0 end,
122
-
reposts = reposts + case when :is_repost then 1 else 0 end,
123
-
quoteposts = quoteposts + case when :is_quotepost then 1 else 0 end
124
-
""", params)
125
-
126
-
def delete_old_posts(self):
127
-
self.db_cnx.execute(self.DELETE_OLD_POSTS_QUERY)
128
-
129
-
def commit_changes(self):
130
-
self.logger.debug('committing changes')
131
-
self.delete_old_posts()
132
-
self.transaction_commit(self.db_cnx)
133
-
self.wal_checkpoint(self.db_cnx, 'RESTART')
134
-
135
-
def serve_feed(self, limit, offset, langs):
136
-
cur = self.db_cnx.execute(self.SERVE_FEED_QUERY, dict(limit=limit, offset=offset))
137
-
return [row[0] for row in cur]
138
-
139
-
def serve_feed_debug(self, limit, offset, langs):
140
-
bindings = dict(limit=limit, offset=offset)
141
-
return apsw.ext.format_query_table(
142
-
self.db_cnx, self.SERVE_FEED_QUERY, bindings,
143
-
string_sanitize=2, text_width=9999, use_unicode=True
144
-
)
-14
feedweb_utils.py
-14
feedweb_utils.py
···
1
-
NGROK_HOSTNAME = 'routinely-right-barnacle.ngrok-free.app'
2
-
3
-
def did_doc():
4
-
return {
5
-
'@context': ['https://www.w3.org/ns/did/v1'],
6
-
'id': f'did:web:{NGROK_HOSTNAME}',
7
-
'service': [
8
-
{
9
-
'id': '#bsky_fg',
10
-
'type': 'BskyFeedGenerator',
11
-
'serviceEndpoint': f'https://{NGROK_HOSTNAME}',
12
-
},
13
-
],
14
-
}
+102
cmd/plc-activity/main.go
+102
cmd/plc-activity/main.go
···
1
+
package main
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"io"
7
+
"log"
8
+
"net/http"
9
+
"strings"
10
+
"time"
11
+
12
+
"github.com/bluesky-social/indigo/atproto/syntax"
13
+
"github.com/redis/go-redis/v9"
14
+
)
15
+
16
+
const PlcExportUrl = `https://plc.directory/export`
17
+
const PlcOpsCountKey = `dev.edavis.muninsky.plc_ops`
18
+
19
+
func main() {
20
+
ctx := context.Background()
21
+
client := http.DefaultClient
22
+
23
+
rdb := redis.NewClient(&redis.Options{
24
+
Addr: "localhost:6379",
25
+
Password: "",
26
+
DB: 0,
27
+
})
28
+
29
+
req, err := http.NewRequestWithContext(ctx, "GET", PlcExportUrl, nil)
30
+
if err != nil {
31
+
panic(err)
32
+
}
33
+
34
+
var lastCursor string
35
+
var cursor string
36
+
cursor = syntax.DatetimeNow().String()
37
+
38
+
q := req.URL.Query()
39
+
q.Add("count", "1000")
40
+
req.URL.RawQuery = q.Encode()
41
+
42
+
for {
43
+
q := req.URL.Query()
44
+
if cursor != "" {
45
+
q.Set("after", cursor)
46
+
}
47
+
req.URL.RawQuery = q.Encode()
48
+
49
+
log.Printf("requesting %s\n", req.URL.String())
50
+
resp, err := client.Do(req)
51
+
if err != nil {
52
+
log.Printf("error doing PLC request: %v\n", err)
53
+
}
54
+
if resp.StatusCode != http.StatusOK {
55
+
log.Printf("PLC request failed status=%d\n", resp.StatusCode)
56
+
}
57
+
58
+
respBytes, err := io.ReadAll(resp.Body)
59
+
if err != nil {
60
+
log.Printf("error reading response body: %v\n", err)
61
+
}
62
+
63
+
lines := strings.Split(string(respBytes), "\n")
64
+
if len(lines) == 0 || (len(lines) == 1 && len(lines[0]) == 0) {
65
+
time.Sleep(5 * time.Second)
66
+
continue
67
+
}
68
+
69
+
var opCount int64
70
+
for _, l := range lines {
71
+
if len(l) < 2 {
72
+
break
73
+
}
74
+
75
+
var op map[string]interface{}
76
+
err = json.Unmarshal([]byte(l), &op)
77
+
if err != nil {
78
+
log.Printf("error decoding JSON: %v\n", err)
79
+
}
80
+
81
+
var ok bool
82
+
cursor, ok = op["createdAt"].(string)
83
+
if !ok {
84
+
log.Printf("missing createdAt")
85
+
}
86
+
87
+
if cursor == lastCursor {
88
+
continue
89
+
}
90
+
91
+
opCount += 1
92
+
lastCursor = cursor
93
+
}
94
+
95
+
log.Printf("fetched %d operations", opCount)
96
+
if _, err := rdb.IncrBy(ctx, PlcOpsCountKey, opCount).Result(); err != nil {
97
+
log.Printf("error incrementing op count in redis: %v\n", err)
98
+
}
99
+
100
+
time.Sleep(5 * time.Second)
101
+
}
102
+
}
+1
-1
autoposters/bskycharts.py
+1
-1
autoposters/bskycharts.py
+6
-6
service/bsky-activity.service
+6
-6
service/bsky-activity.service
···
1
1
[Unit]
2
-
Description=Bsky Activity
2
+
Description=bsky activity
3
3
After=network.target syslog.target
4
4
5
5
[Service]
6
6
Type=simple
7
-
User=eric
8
-
WorkingDirectory=/home/eric/bsky-tools
9
-
ExecStart=/home/eric/.local/bin/pipenv run ./bsky-activity.py
7
+
User=ubuntu
8
+
WorkingDirectory=/home/ubuntu/bsky-tools
9
+
ExecStart=/home/ubuntu/bsky-tools/bin/bsky-activity
10
10
TimeoutSec=15
11
-
Restart=on-failure
12
-
RestartSec=1
11
+
Restart=always
12
+
RestartSec=30
13
13
StandardOutput=journal
14
14
15
15
[Install]
+55
-8
cmd/bsky-activity/main.go
+55
-8
cmd/bsky-activity/main.go
···
7
7
"os"
8
8
"os/signal"
9
9
"strings"
10
+
"sync"
10
11
"syscall"
12
+
"time"
11
13
12
14
appbsky "github.com/bluesky-social/indigo/api/bsky"
13
15
jetstream "github.com/bluesky-social/jetstream/pkg/models"
···
15
17
"github.com/redis/go-redis/v9"
16
18
)
17
19
20
+
type Queue struct {
21
+
lk sync.Mutex
22
+
events []jetstream.Event
23
+
}
24
+
25
+
func NewQueue(capacity int) *Queue {
26
+
return &Queue{
27
+
events: make([]jetstream.Event, 0, capacity),
28
+
}
29
+
}
30
+
31
+
func (q *Queue) Enqueue(event jetstream.Event) {
32
+
q.lk.Lock()
33
+
defer q.lk.Unlock()
34
+
35
+
q.events = append(q.events, event)
36
+
}
37
+
38
+
func (q *Queue) Dequeue() (jetstream.Event, bool) {
39
+
q.lk.Lock()
40
+
defer q.lk.Unlock()
41
+
42
+
var event jetstream.Event
43
+
44
+
if len(q.events) == 0 {
45
+
return event, false
46
+
}
47
+
48
+
event = q.events[0]
49
+
q.events = q.events[1:]
50
+
return event, true
51
+
}
52
+
53
+
func (q *Queue) Size() int {
54
+
q.lk.Lock()
55
+
defer q.lk.Unlock()
56
+
57
+
return len(q.events)
58
+
}
59
+
18
60
const JetstreamUrl = `wss://jetstream1.us-west.bsky.network/subscribe`
19
61
20
62
var AppBskyAllowlist = map[string]bool{
···
59
101
return false
60
102
}
61
103
62
-
func handler(ctx context.Context, events <-chan jetstream.Event) {
104
+
func handler(ctx context.Context, queue *Queue) {
63
105
rdb := redis.NewClient(&redis.Options{
64
106
Addr: "localhost:6379",
65
107
Password: "",
···
69
111
var eventCount int
70
112
71
113
eventLoop:
72
-
for event := range events {
114
+
for {
73
115
select {
74
116
case <-ctx.Done():
75
117
break eventLoop
76
118
default:
77
119
}
78
120
121
+
event, ok := queue.Dequeue()
122
+
if !ok {
123
+
time.Sleep(100 * time.Millisecond)
124
+
continue
125
+
}
126
+
79
127
if event.Kind != jetstream.EventKindCommit {
80
128
continue
81
129
}
···
145
193
if _, err := pipe.Exec(ctx); err != nil {
146
194
log.Printf("failed to exec pipe\n")
147
195
}
196
+
log.Printf("queue size: %d\n", queue.Size())
148
197
}
149
198
}
150
199
}
···
164
213
log.Printf("websocket closed\n")
165
214
}()
166
215
167
-
jetstreamEvents := make(chan jetstream.Event)
168
-
go handler(ctx, jetstreamEvents)
216
+
queue := NewQueue(100_000)
217
+
go handler(ctx, queue)
169
218
170
219
log.Printf("starting up\n")
171
-
var event jetstream.Event
172
220
go func() {
173
221
for {
174
-
event = jetstream.Event{}
222
+
var event jetstream.Event
175
223
err := conn.ReadJSON(&event)
176
224
if err != nil {
177
225
log.Printf("ReadJSON error: %v\n", err)
178
226
stop()
179
227
break
180
-
} else {
181
-
jetstreamEvents <- event
182
228
}
229
+
queue.Enqueue(event)
183
230
}
184
231
}()
185
232
+11
-3
go.mod
+11
-3
go.mod
···
1
1
module github.com/edavis/bsky-tools
2
2
3
-
go 1.23.0
3
+
go 1.24
4
+
5
+
toolchain go1.24.7
4
6
5
7
require (
6
-
github.com/bluesky-social/indigo v0.0.0-20240905024844-a4f38639767f
8
+
github.com/bluesky-social/indigo v0.0.0-20250909204019-c5eaa30f683f
7
9
github.com/bluesky-social/jetstream v0.0.0-20241020000921-dcd43344c716
10
+
github.com/fxamacker/cbor/v2 v2.9.0
8
11
github.com/gorilla/websocket v1.5.1
9
12
github.com/mattn/go-sqlite3 v1.14.22
10
13
github.com/pemistahl/lingua-go v1.4.0
11
14
github.com/redis/go-redis/v9 v9.3.0
15
+
github.com/urfave/cli/v2 v2.26.0
12
16
)
13
17
14
18
require (
15
19
github.com/carlmjohnson/versioninfo v0.22.5 // indirect
16
20
github.com/cespare/xxhash/v2 v2.3.0 // indirect
21
+
github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect
17
22
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
18
23
github.com/felixge/httpsnoop v1.0.4 // indirect
19
24
github.com/go-logr/logr v1.4.1 // indirect
···
48
53
github.com/multiformats/go-varint v0.0.7 // indirect
49
54
github.com/opentracing/opentracing-go v1.2.0 // indirect
50
55
github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect
56
+
github.com/russross/blackfriday/v2 v2.1.0 // indirect
51
57
github.com/shopspring/decimal v1.3.1 // indirect
52
58
github.com/spaolacci/murmur3 v1.1.0 // indirect
53
-
github.com/whyrusleeping/cbor-gen v0.1.3-0.20240904181319-8dc02b38228c // indirect
59
+
github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e // indirect
60
+
github.com/x448/float16 v0.8.4 // indirect
61
+
github.com/xrash/smetrics v0.0.0-20231213231151-1d8dd44e695e // indirect
54
62
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
55
63
go.opentelemetry.io/otel v1.21.0 // indirect
56
64
go.opentelemetry.io/otel/metric v1.21.0 // indirect
+16
-4
go.sum
+16
-4
go.sum
···
1
1
github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
2
2
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
3
-
github.com/bluesky-social/indigo v0.0.0-20240905024844-a4f38639767f h1:Q9cfCAlYWIWPsSDhg5w6qcutQ7YaJtfTjiRLP/mw+pc=
4
-
github.com/bluesky-social/indigo v0.0.0-20240905024844-a4f38639767f/go.mod h1:Zx9nSWgd/FxMenkJW07VKnzspxpHBdPrPmS+Fspl2I0=
3
+
github.com/bluesky-social/indigo v0.0.0-20250909204019-c5eaa30f683f h1:FugOoTzh0nCMTWGqNGsjttFWVPcwxaaGD3p/nE9V8qY=
4
+
github.com/bluesky-social/indigo v0.0.0-20250909204019-c5eaa30f683f/go.mod h1:n6QE1NDPFoi7PRbMUZmc2y7FibCqiVU4ePpsvhHUBR8=
5
5
github.com/bluesky-social/jetstream v0.0.0-20241020000921-dcd43344c716 h1:I8+VaZKaNIGCPGXE2/VXzJGlPFEZgiFLjnge+OWFl5w=
6
6
github.com/bluesky-social/jetstream v0.0.0-20241020000921-dcd43344c716/go.mod h1:/dE2dmFell/m4zxgIbH3fkiqZ1obzr/ETj4RpgomgMs=
7
7
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
···
13
13
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
14
14
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
15
15
github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
16
+
github.com/cpuguy83/go-md2man/v2 v2.0.3 h1:qMCsGGgs+MAzDFyp9LpAe1Lqy/fY/qCovCm0qnXZOBM=
17
+
github.com/cpuguy83/go-md2man/v2 v2.0.3/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
16
18
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
17
19
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
18
20
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
···
20
22
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
21
23
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
22
24
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
25
+
github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM=
26
+
github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ=
23
27
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
24
28
github.com/go-logr/logr v1.4.1 h1:pKouT5E8xu9zeFC39JXRDukb6JFQPXM5p5I91188VAQ=
25
29
github.com/go-logr/logr v1.4.1/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
···
124
128
github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
125
129
github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
126
130
github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
131
+
github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
132
+
github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
127
133
github.com/shopspring/decimal v1.3.1 h1:2Usl1nmF/WZucqkFZhnfFYxxxu8LG21F6nPQBE5gKV8=
128
134
github.com/shopspring/decimal v1.3.1/go.mod h1:DKyhrW/HYNuLGql+MJL6WCR6knT2jwCFRcu2hWCYk4o=
129
135
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
···
141
147
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
142
148
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
143
149
github.com/urfave/cli v1.22.10/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
150
+
github.com/urfave/cli/v2 v2.26.0 h1:3f3AMg3HpThFNT4I++TKOejZO8yU55t3JnnSr4S4QEI=
151
+
github.com/urfave/cli/v2 v2.26.0/go.mod h1:8qnjx1vcq5s2/wpsqoZFndg2CE5tNFyrTvS6SinrnYQ=
144
152
github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0 h1:GDDkbFiaK8jsSDJfjId/PEGEShv6ugrt4kYsC5UIDaQ=
145
153
github.com/warpfork/go-wish v0.0.0-20220906213052-39a1cc7a02d0/go.mod h1:x6AKhvSSexNrVSrViXSHUEbICjmGXhtgABaHIySUSGw=
146
-
github.com/whyrusleeping/cbor-gen v0.1.3-0.20240904181319-8dc02b38228c h1:UsxJNcLPfyLyVaA4iusIrsLAqJn/xh36Qgb8emqtXzk=
147
-
github.com/whyrusleeping/cbor-gen v0.1.3-0.20240904181319-8dc02b38228c/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so=
154
+
github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e h1:28X54ciEwwUxyHn9yrZfl5ojgF4CBNLWX7LR0rvBkf4=
155
+
github.com/whyrusleeping/cbor-gen v0.2.1-0.20241030202151-b7a6831be65e/go.mod h1:pM99HXyEbSQHcosHc0iW7YFmwnscr+t9Te4ibko05so=
156
+
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
157
+
github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
158
+
github.com/xrash/smetrics v0.0.0-20231213231151-1d8dd44e695e h1:+SOyEddqYF09QP7vr7CgJ1eti3pY9Fn3LHO1M1r/0sI=
159
+
github.com/xrash/smetrics v0.0.0-20231213231151-1d8dd44e695e/go.mod h1:N3UwUGtsrSj3ccvlPHLoLsHnpR27oXr4ZE984MbSER8=
148
160
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
149
161
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
150
162
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
+24
cmd/bsky-modactions/main.go
+24
cmd/bsky-modactions/main.go
···
8
8
"net/http"
9
9
"os"
10
10
"os/signal"
11
+
"time"
11
12
12
13
comatproto "github.com/bluesky-social/indigo/api/atproto"
13
14
"github.com/fxamacker/cbor/v2"
···
102
103
}
103
104
}()
104
105
106
+
mux := http.NewServeMux()
107
+
mux.HandleFunc("/config", configHandler)
108
+
mux.HandleFunc("/", valueHandler)
109
+
110
+
srv := &http.Server{
111
+
Addr: "127.0.0.1:4456",
112
+
Handler: mux,
113
+
}
114
+
115
+
go func() {
116
+
if err := srv.ListenAndServe(); err != nil {
117
+
slog.Error("error starting HTTP server", "err", err)
118
+
return
119
+
}
120
+
}()
121
+
105
122
<-ctx.Done()
106
123
stop()
107
124
slog.Info("shutting down")
108
125
126
+
endctx, cancel := context.WithTimeout(context.TODO(), time.Minute)
127
+
defer cancel()
128
+
129
+
if err := srv.Shutdown(endctx); err != nil {
130
+
slog.Error("error shutting down server", "err", err)
131
+
}
132
+
109
133
return nil
110
134
}
111
135