bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

communication: make pin job mutable

Browse Source
This commit is contained in:
Brad Stein 2025-12-31 15:23:17 -03:00
parent d8d741bbd9
commit 6203faae3f
1 changed files with 97 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

153
services/communication/pin-othrys-job.yaml
View File

@ -1,68 +1,109 @@
# services/communication/pin-othrys-job.yaml # services/communication/pin-othrys-job.yaml
apiVersion: batch/v1 apiVersion: batch/v1
kind: Job kind: CronJob
metadata: metadata:
name: pin-othrys-invite name: pin-othrys-invite
namespace: communication namespace: communication
spec: spec:
ttlSecondsAfterFinished: 3600 schedule: "*/30 * * * *"
template: concurrencyPolicy: Forbid
successfulJobsHistoryLimit: 1
failedJobsHistoryLimit: 1
jobTemplate:
spec: spec:
restartPolicy: OnFailure template:
containers: spec:
- name: pin restartPolicy: OnFailure
image: python:3.11-slim containers:
env: - name: pin
- name: SYNAPSE_BASE image: python:3.11-slim
value: http://othrys-synapse-matrix-synapse:8008 env:
- name: SEEDER_USER - name: SYNAPSE_BASE
value: othrys-seeder value: http://othrys-synapse-matrix-synapse:8008
- name: SEEDER_PASS - name: SEEDER_USER
valueFrom: value: othrys-seeder
secretKeyRef: - name: SEEDER_PASS
name: atlasbot-credentials-runtime valueFrom:
key: seeder-password secretKeyRef:
command: name: atlasbot-credentials-runtime
- /bin/sh key: seeder-password
- -c command:
- | - /bin/sh
set -euo pipefail - -c
pip install --no-cache-dir requests >/dev/null - |
python - <<'PY' set -euo pipefail
import requests, urllib.parse, os pip install --no-cache-dir requests >/dev/null
python - <<'PY'
import os, requests, urllib.parse
BASE = os.environ["SYNAPSE_BASE"] BASE = os.environ["SYNAPSE_BASE"]
def login(user, password): ROOM_ALIAS = "#othrys:live.bstein.dev"
r = requests.post(f"{BASE}/_matrix/client/v3/login", json={ MESSAGE = (
"type": "m.login.password", "Invite guests: share https://live.bstein.dev/#/room/#othrys:live.bstein.dev?action=join "
"identifier": {"type": "m.id.user", "user": user}, "and choose 'Continue' -> 'Join as guest'."
"password": password, )
})
r.raise_for_status()
return r.json()["access_token"]
def resolve(alias, token): def auth(token): return {"Authorization": f"Bearer {token}"}
enc = urllib.parse.quote(alias)
r = requests.get(f"{BASE}/_matrix/client/v3/directory/room/{enc}", headers={"Authorization": f"Bearer {token}"})
r.raise_for_status()
return r.json()["room_id"]
def send(room_id, token, body): def login(user, password):
r = requests.post(f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/send/m.room.message", r = requests.post(f"{BASE}/_matrix/client/v3/login", json={
headers={"Authorization": f"Bearer {token}"}, "type": "m.login.password",
json={"msgtype": "m.text", "body": body}) "identifier": {"type": "m.id.user", "user": user},
r.raise_for_status() "password": password,
return r.json()["event_id"] })
r.raise_for_status()
return r.json()["access_token"]
def pin(room_id, token, event_id): def resolve(alias, token):
r = requests.put(f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/state/m.room.pinned_events", enc = urllib.parse.quote(alias)
headers={"Authorization": f"Bearer {token}"}, r = requests.get(f"{BASE}/_matrix/client/v3/directory/room/{enc}", headers=auth(token))
json={"pinned": [event_id]}) r.raise_for_status()
r.raise_for_status() return r.json()["room_id"]
token = login(os.environ["SEEDER_USER"], os.environ["SEEDER_PASS"]) def get_pinned(room_id, token):
room_id = resolve("#othrys:live.bstein.dev", token) r = requests.get(
msg = "Invite guests: share https://live.bstein.dev/#/room/#othrys:live.bstein.dev?action=join and choose 'Continue' -> 'Join as guest'." f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/state/m.room.pinned_events",
eid = send(room_id, token, msg) headers=auth(token),
pin(room_id, token, eid) )
PY if r.status_code == 404:
return []
r.raise_for_status()
return r.json().get("pinned", [])
def get_event(room_id, event_id, token):
r = requests.get(
f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/event/{urllib.parse.quote(event_id)}",
headers=auth(token),
)
if r.status_code == 404:
return None
r.raise_for_status()
return r.json()
def send(room_id, token, body):
r = requests.post(
f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/send/m.room.message",
headers=auth(token),
json={"msgtype": "m.text", "body": body},
)
r.raise_for_status()
return r.json()["event_id"]
def pin(room_id, token, event_id):
r = requests.put(
f"{BASE}/_matrix/client/v3/rooms/{urllib.parse.quote(room_id)}/state/m.room.pinned_events",
headers=auth(token),
json={"pinned": [event_id]},
)
r.raise_for_status()
token = login(os.environ["SEEDER_USER"], os.environ["SEEDER_PASS"])
room_id = resolve(ROOM_ALIAS, token)
for event_id in get_pinned(room_id, token):
ev = get_event(room_id, event_id, token)
if ev and ev.get("content", {}).get("body") == MESSAGE:
raise SystemExit(0)
eid = send(room_id, token, MESSAGE)
pin(room_id, token, eid)
PY