mirror of
https://github.com/believethehype/nostrdvm.git
synced 2025-03-17 13:21:48 +01:00
988 lines
56 KiB
Python
988 lines
56 KiB
Python
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
from sys import platform
|
|
|
|
from nostr_sdk import PublicKey, Keys, Client, Tag, Event, EventBuilder, Filter, HandleNotification, Timestamp, \
|
|
LogLevel, Options, nip04_encrypt, nip44_encrypt, Nip44Version, Kind, RelayLimits, uniffi_set_event_loop, ClientBuilder, NostrSigner
|
|
|
|
|
|
from nostr_dvm.utils.admin_utils import admin_make_database_updates, AdminConfig
|
|
from nostr_dvm.utils.backend_utils import get_amount_per_task, check_task_is_supported, get_task
|
|
from nostr_dvm.utils.cashu_utils import redeem_cashu
|
|
from nostr_dvm.utils.database_utils import create_sql_table, get_or_add_user, update_sql_table, \
|
|
update_user_subscription
|
|
from nostr_dvm.utils.definitions import EventDefinitions, RequiredJobToWatch, JobToWatch
|
|
from nostr_dvm.utils.dvmconfig import DVMConfig
|
|
from nostr_dvm.utils.mediasource_utils import input_data_file_duration
|
|
from nostr_dvm.utils.nip88_utils import nip88_has_active_subscription
|
|
from nostr_dvm.utils.nostr_utils import get_event_by_id, get_referenced_event_by_id, check_and_decrypt_tags, \
|
|
send_event_outbox, print_send_result
|
|
from nostr_dvm.utils.nut_wallet_utils import NutZapWallet
|
|
from nostr_dvm.utils.output_utils import build_status_reaction
|
|
from nostr_dvm.utils.print_utils import bcolors
|
|
from nostr_dvm.utils.zap_utils import check_bolt11_ln_bits_is_paid, create_bolt11_ln_bits, parse_zap_event_tags, \
|
|
parse_amount_from_bolt11_invoice, zaprequest, pay_bolt11_ln_bits, create_bolt11_lud16
|
|
|
|
#os.environ["RUST_BACKTRACE"] = "full"
|
|
|
|
|
|
|
|
class DVM:
|
|
dvm_config: DVMConfig
|
|
admin_config: AdminConfig
|
|
keys: Keys
|
|
client: Client
|
|
job_list: list
|
|
jobs_on_hold_list: list
|
|
stop_thread = False
|
|
|
|
def __init__(self, dvm_config, admin_config=None, stop_thread=False):
|
|
try:
|
|
asyncio.run(self.run_dvm(dvm_config, admin_config, stop_thread))
|
|
#uniffi_set_event_loop(asyncio.get_running_loop())
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
|
|
|
|
|
|
async def run_dvm(self, dvm_config, admin_config, stop_thread):
|
|
self.dvm_config = dvm_config
|
|
self.admin_config = admin_config
|
|
self.keys = Keys.parse(dvm_config.PRIVATE_KEY)
|
|
relaylimits = RelayLimits.disable()
|
|
opts = Options().relay_limits(relaylimits) #.difficulty(28)
|
|
|
|
#self.client = Client(self.keys)
|
|
self.client = ClientBuilder().signer(NostrSigner.keys(self.keys)).opts(opts).build()
|
|
self.job_list = []
|
|
self.jobs_on_hold_list = []
|
|
pk = self.keys.public_key()
|
|
print(bcolors.BLUE + "[" + self.dvm_config.NIP89.NAME + "] " + "Nostr DVM public key: " + str(
|
|
pk.to_bech32()) + " Hex: " +
|
|
str(pk.to_hex()) + " Supported DVM tasks: " +
|
|
', '.join(p.NAME + ":" + p.TASK for p in self.dvm_config.SUPPORTED_DVMS) + bcolors.ENDC)
|
|
|
|
for relay in self.dvm_config.RELAY_LIST:
|
|
await self.client.add_relay(relay)
|
|
await self.client.connect()
|
|
|
|
zap_filter = Filter().pubkey(pk).kinds([EventDefinitions.KIND_ZAP, EventDefinitions.KIND_NIP61_NUT_ZAP]).since(
|
|
Timestamp.now())
|
|
kinds = [EventDefinitions.KIND_NIP90_GENERIC]
|
|
for dvm in self.dvm_config.SUPPORTED_DVMS:
|
|
if dvm.KIND not in kinds:
|
|
kinds.append(dvm.KIND)
|
|
dvm_filter = (Filter().kinds(kinds).since(Timestamp.now()))
|
|
ping_filter= Filter().pubkey(pk).kind(EventDefinitions.KIND_NIP90_PING).since(Timestamp.now())
|
|
create_sql_table(self.dvm_config.DB)
|
|
await admin_make_database_updates(adminconfig=self.admin_config, dvmconfig=self.dvm_config, client=self.client)
|
|
await self.client.subscribe(dvm_filter, None)
|
|
await self.client.subscribe(zap_filter, None)
|
|
await self.client.subscribe(ping_filter, None)
|
|
|
|
if self.dvm_config.ENABLE_NUTZAP:
|
|
nutzap_wallet = NutZapWallet()
|
|
nut_wallet = await nutzap_wallet.get_nut_wallet(self.client, self.keys)
|
|
|
|
if nut_wallet is None:
|
|
await nutzap_wallet.create_new_nut_wallet(self.dvm_config.NUZAP_MINTS, self.dvm_config.NUTZAP_RELAYS,
|
|
self.client, self.keys, "DVM", "DVM Nutsack")
|
|
nut_wallet = await nutzap_wallet.get_nut_wallet(self.client, self.keys)
|
|
|
|
await nutzap_wallet.announce_nutzap_info_event(nut_wallet, self.client, self.keys)
|
|
|
|
if self.dvm_config.REANNOUNCE_MINTS:
|
|
nut_wallet.mints = dvm_config.NUZAP_MINTS
|
|
await nutzap_wallet.announce_nutzap_info_event(nut_wallet, self.client, self.keys)
|
|
|
|
class NotificationHandler(HandleNotification):
|
|
client = self.client
|
|
dvm_config = self.dvm_config
|
|
keys = self.keys
|
|
|
|
async def handle(self, relay_url, subscription_id, nostr_event: Event):
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print(nostr_event.as_json())
|
|
if EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u16() <= nostr_event.kind().as_u16() <= EventDefinitions.KIND_NIP90_GENERIC.as_u16():
|
|
await handle_nip90_job_event(nostr_event)
|
|
elif nostr_event.kind().as_u16() == EventDefinitions.KIND_ZAP.as_u16():
|
|
await handle_zap(nostr_event)
|
|
elif nostr_event.kind().as_u16() == EventDefinitions.KIND_NIP61_NUT_ZAP.as_u16():
|
|
await handle_nutzap(nostr_event)
|
|
elif nostr_event.kind().as_u16() == EventDefinitions.KIND_NIP90_PING.as_u16():
|
|
await handle_ping(nostr_event)
|
|
|
|
async def handle_msg(self, relay_url, msg):
|
|
return
|
|
|
|
async def handle_nip90_job_event(nip90_event):
|
|
# decrypted encrypted events
|
|
|
|
nip90_event, use_legacy_encryption = check_and_decrypt_tags(nip90_event, self.dvm_config)
|
|
# if event is encrypted, but we can't decrypt it (e.g. because its directed to someone else), return
|
|
if nip90_event is None:
|
|
return
|
|
|
|
task_is_free = False
|
|
user_has_active_subscription = False
|
|
cashu = ""
|
|
p_tag_str = ""
|
|
|
|
for tag in nip90_event.tags().to_vec():
|
|
if tag.as_vec()[0] == "cashu":
|
|
cashu = tag.as_vec()[1]
|
|
elif tag.as_vec()[0] == "p":
|
|
p_tag_str = tag.as_vec()[1]
|
|
|
|
if p_tag_str != "" and p_tag_str != self.dvm_config.PUBLIC_KEY:
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] No public request, also not addressed to me.")
|
|
return
|
|
|
|
# check if task is supported by the current DVM
|
|
task_supported, task = await check_task_is_supported(nip90_event, client=self.client,
|
|
config=self.dvm_config)
|
|
# if task is supported, continue, else do nothing.
|
|
if task_supported:
|
|
# fetch or add user contacting the DVM from/to local database
|
|
user = await get_or_add_user(self.dvm_config.DB, nip90_event.author().to_hex(), client=self.client,
|
|
config=self.dvm_config, skip_meta=False)
|
|
# if user is blacklisted for some reason, send an error reaction and return
|
|
if user.isblacklisted:
|
|
# await send_job_status_reaction(nip90_event, "error", client=self.client, dvm_config=self.dvm_config)
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Request by blacklisted user, skipped")
|
|
return
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.INFO.value:
|
|
print(
|
|
bcolors.MAGENTA + "[" + self.dvm_config.NIP89.NAME + "] Received new Request: " + task + " from " + user.name + " (" + PublicKey.parse(
|
|
user.npub).to_bech32() + ")" + bcolors.ENDC)
|
|
duration = await input_data_file_duration(nip90_event, dvm_config=self.dvm_config, client=self.client)
|
|
amount = get_amount_per_task(task, self.dvm_config, duration)
|
|
if amount is None:
|
|
return
|
|
|
|
# If this is a subscription DVM and the Task is directed to us, check for active subscription
|
|
if dvm_config.NIP88 is not None and p_tag_str == self.dvm_config.PUBLIC_KEY:
|
|
# await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
|
|
# "Checking Subscription Status, please wait..", self.dvm_config)
|
|
# if we stored in the database that the user has an active subscription, we don't need to check it
|
|
print("User Subscription: " + str(user.subscribed) + " Current time: " + str(
|
|
Timestamp.now().as_secs()))
|
|
# if we have an entry in the db that user is subscribed, continue
|
|
if int(user.subscribed) > int(Timestamp.now().as_secs()):
|
|
print("User subscribed until: " + str(Timestamp.from_secs(user.subscribed).to_human_datetime()))
|
|
user_has_active_subscription = True
|
|
await send_job_status_reaction(nip90_event, "subscription-active", True, amount,
|
|
self.client, "User subscripton active until " +
|
|
Timestamp.from_secs(
|
|
int(user.subscribed)).to_human_datetime().replace(
|
|
"Z", " ").replace("T", " ") + " GMT", self.dvm_config)
|
|
# otherwise we check for an active subscription by checking recipie events
|
|
# sleep a little to not get rate limited
|
|
await asyncio.sleep(0.5)
|
|
|
|
else:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Checking Subscription status")
|
|
# await send_job_status_reaction(nip90_event, "subscription-required", True, amount, self.client,
|
|
# "I Don't have information about subscription status, checking on the Nostr. This might take a few seconds",
|
|
# self.dvm_config)
|
|
|
|
subscription_status = await nip88_has_active_subscription(PublicKey.parse(user.npub),
|
|
self.dvm_config.NIP88.DTAG,
|
|
self.client,
|
|
self.dvm_config.PUBLIC_KEY)
|
|
|
|
if subscription_status["isActive"]:
|
|
await send_job_status_reaction(nip90_event, "subscription-required", True, amount,
|
|
self.client,
|
|
"User subscripton active until " + Timestamp.from_secs(int(
|
|
subscription_status[
|
|
"validUntil"])).to_human_datetime().replace("Z",
|
|
" ").replace(
|
|
"T", " ") + " GMT",
|
|
self.dvm_config)
|
|
|
|
print("Checked Recipe: User subscribed until: " + str(
|
|
Timestamp.from_secs(int(subscription_status["validUntil"])).to_human_datetime()))
|
|
user_has_active_subscription = True
|
|
update_user_subscription(user.npub,
|
|
int(subscription_status["validUntil"]),
|
|
self.client, self.dvm_config)
|
|
|
|
# sleep a little before sending next status update
|
|
|
|
|
|
else:
|
|
print("No active subscription found")
|
|
await send_job_status_reaction(nip90_event, "subscription-required", True, amount,
|
|
self.client,
|
|
"No active subscription found. Manage your subscription at: " + self.dvm_config.SUBSCRIPTION_MANAGEMENT,
|
|
self.dvm_config)
|
|
|
|
for dvm in self.dvm_config.SUPPORTED_DVMS:
|
|
if (
|
|
dvm.TASK == task or dvm.TASK == "generic") and dvm.FIX_COST == 0 and dvm.PER_UNIT_COST == 0 and dvm_config.NIP88 is None:
|
|
task_is_free = True
|
|
|
|
cashu_redeemed = False
|
|
if cashu != "":
|
|
print(cashu)
|
|
cashu_redeemed, cashu_message, redeem_amount, fees = await redeem_cashu(cashu, self.dvm_config,
|
|
self.client, int(amount))
|
|
print(cashu_message)
|
|
if cashu_message != "success":
|
|
await send_job_status_reaction(nip90_event, "error", False, amount, self.client, cashu_message,
|
|
self.dvm_config)
|
|
return
|
|
# if user is whitelisted or task is free, just dxo the job
|
|
if (user.iswhitelisted or task_is_free or cashu_redeemed) and (
|
|
p_tag_str == "" or p_tag_str ==
|
|
self.dvm_config.PUBLIC_KEY):
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print(
|
|
"[" + self.dvm_config.NIP89.NAME + "] Free task or Whitelisted for task " + task +
|
|
". Starting processing..")
|
|
|
|
if dvm_config.SEND_FEEDBACK_EVENTS:
|
|
await send_job_status_reaction(nip90_event, "processing", True, 0,
|
|
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
|
|
client=self.client, dvm_config=self.dvm_config)
|
|
|
|
# when we reimburse users on error make sure to not send anything if it was free
|
|
if user.iswhitelisted or task_is_free:
|
|
amount = 0
|
|
await do_work(nip90_event, amount)
|
|
# if task is directed to us via p tag and user has balance or is subscribed, do the job and update balance
|
|
elif (p_tag_str == self.dvm_config.PUBLIC_KEY and (
|
|
user.balance >= int(
|
|
amount) and dvm_config.NIP88 is None) or (
|
|
p_tag_str == self.dvm_config.PUBLIC_KEY and user_has_active_subscription)):
|
|
|
|
if not user_has_active_subscription:
|
|
balance = max(user.balance - int(amount), 0)
|
|
update_sql_table(db=self.dvm_config.DB, npub=user.npub, balance=balance,
|
|
iswhitelisted=user.iswhitelisted, isblacklisted=user.isblacklisted,
|
|
nip05=user.nip05, lud16=user.lud16, name=user.name,
|
|
lastactive=Timestamp.now().as_secs(), subscribed=user.subscribed)
|
|
|
|
print(
|
|
"[" + self.dvm_config.NIP89.NAME + "] Using user's balance for task: " + task +
|
|
". Starting processing.. New balance is: " + str(balance))
|
|
else:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] User has active subscription for task: " + task +
|
|
". Starting processing.. Balance remains at: " + str(user.balance))
|
|
|
|
await send_job_status_reaction(nip90_event, "processing", True, 0,
|
|
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
|
|
client=self.client, dvm_config=self.dvm_config)
|
|
|
|
await do_work(nip90_event, amount)
|
|
|
|
# else send a payment required event to user
|
|
elif p_tag_str == "" or p_tag_str == self.dvm_config.PUBLIC_KEY:
|
|
|
|
if dvm_config.NIP88 is not None:
|
|
print(
|
|
"[" + self.dvm_config.NIP89.NAME + "] Hinting user for Subscription: " +
|
|
nip90_event.id().to_hex())
|
|
# await send_job_status_reaction(nip90_event, "subscription-required",
|
|
# False, 0, client=self.client,
|
|
# dvm_config=self.dvm_config)
|
|
else:
|
|
bid = 0
|
|
for tag in nip90_event.tags().to_vec():
|
|
if tag.as_vec()[0] == 'bid':
|
|
bid = int(tag.as_vec()[1])
|
|
|
|
print(
|
|
"[" + self.dvm_config.NIP89.NAME + "] Payment required: New Nostr " + task + " Job event: "
|
|
+ nip90_event.as_json())
|
|
if bid > 0:
|
|
bid_offer = int(bid / 1000)
|
|
if bid_offer >= int(amount):
|
|
await send_job_status_reaction(nip90_event, "payment-required", False,
|
|
int(amount), # bid_offer
|
|
client=self.client, dvm_config=self.dvm_config)
|
|
|
|
else: # If there is no bid, just request server rate from user
|
|
print(
|
|
"[" + self.dvm_config.NIP89.NAME + "] Requesting payment for Event: " +
|
|
nip90_event.id().to_hex())
|
|
await send_job_status_reaction(nip90_event, "payment-required",
|
|
False, int(amount), client=self.client,
|
|
dvm_config=self.dvm_config)
|
|
|
|
|
|
|
|
|
|
else:
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Job addressed to someone else, skipping..")
|
|
# else:
|
|
# print("[" + self.dvm_config.NIP89.NAME + "] Task " + task + " not supported on this DVM, skipping..")
|
|
|
|
|
|
async def handle_ping(event):
|
|
nip90_event, use_legacy_encryption = check_and_decrypt_tags(event, self.dvm_config)
|
|
# if event is encrypted, but we can't decrypt it (e.g. because its directed to someone else), return
|
|
if nip90_event is None:
|
|
return
|
|
|
|
await send_job_status_reaction(nip90_event, "pong", client=self.client,
|
|
content="alive and kicking",
|
|
dvm_config=self.dvm_config)
|
|
print("Sent pong")
|
|
|
|
|
|
async def handle_nutzap(nut_zap_event):
|
|
if self.dvm_config.ENABLE_NUTZAP:
|
|
nut_wallet = await nutzap_wallet.get_nut_wallet(self.client, self.keys)
|
|
if nut_wallet is not None:
|
|
received_amount, message, sender = await nutzap_wallet.reedeem_nutzap(nut_zap_event, nut_wallet,
|
|
self.client, self.keys)
|
|
user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client,
|
|
config=self.dvm_config)
|
|
zapped_event = None
|
|
for tag in nut_zap_event.tags().to_vec():
|
|
if tag.as_vec()[0] == 'e':
|
|
zapped_event = await get_event_by_id(tag.as_vec()[1], client=self.client,
|
|
config=self.dvm_config)
|
|
|
|
if zapped_event is not None:
|
|
if zapped_event.kind() == EventDefinitions.KIND_FEEDBACK:
|
|
amount = 0
|
|
job_event = None
|
|
p_tag_str = ""
|
|
status = ""
|
|
for tag in zapped_event.tags().to_vec():
|
|
if tag.as_vec()[0] == 'amount':
|
|
amount = int(float(tag.as_vec()[1]) / 1000)
|
|
elif tag.as_vec()[0] == 'e':
|
|
job_event = await get_event_by_id(tag.as_vec()[1], client=self.client,
|
|
config=self.dvm_config)
|
|
if job_event is not None:
|
|
job_event, use_legacy_encryption = check_and_decrypt_tags(job_event, self.dvm_config)
|
|
if job_event is None:
|
|
return
|
|
else:
|
|
return
|
|
elif tag.as_vec()[0] == 'status':
|
|
status = tag.as_vec()[1]
|
|
|
|
# if a reaction by us got zapped
|
|
print(status)
|
|
|
|
task_supported, task = await check_task_is_supported(job_event, client=self.client,
|
|
config=self.dvm_config)
|
|
if job_event is not None and task_supported:
|
|
print("NutZap received for NIP90 task: " + str(received_amount) + " Sats from " + str(
|
|
user.name + " (" + user.npub + ")"))
|
|
if amount <= received_amount:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Payment-request fulfilled...")
|
|
await send_job_status_reaction(job_event, "processing", client=self.client,
|
|
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
|
|
dvm_config=self.dvm_config)
|
|
indices = [i for i, x in enumerate(self.job_list) if
|
|
x.event == job_event]
|
|
index = -1
|
|
if len(indices) > 0:
|
|
index = indices[0]
|
|
if index > -1:
|
|
if self.job_list[index].is_processed:
|
|
self.job_list[index].is_paid = True
|
|
await check_and_return_event(self.job_list[index].result, job_event)
|
|
elif not (self.job_list[index]).is_processed:
|
|
# If payment-required appears before processing
|
|
self.job_list.pop(index)
|
|
print("Starting work...")
|
|
await do_work(job_event, received_amount)
|
|
else:
|
|
print("Job not in List, but starting work...")
|
|
await do_work(job_event, received_amount)
|
|
|
|
else:
|
|
await send_job_status_reaction(job_event, "payment-rejected",
|
|
False, received_amount, client=self.client,
|
|
dvm_config=self.dvm_config)
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Invoice was not paid sufficiently")
|
|
|
|
if self.dvm_config.ENABLE_AUTO_MELT:
|
|
balance = nut_wallet.balance + received_amount
|
|
if balance > self.dvm_config.AUTO_MELT_AMOUNT:
|
|
lud16 = self.admin_config.LUD16
|
|
npub = self.dvm_config.PUBLIC_KEY
|
|
mint_index = 0
|
|
await nutzap_wallet.melt_cashu(nut_wallet, self.dvm_config.NUZAP_MINTS[mint_index],
|
|
self.dvm_config.AUTO_MELT_AMOUNT, self.client, self.keys,
|
|
lud16, npub)
|
|
|
|
else:
|
|
print("NutZaps not enabled for this DVM. ")
|
|
|
|
async def handle_zap(zap_event):
|
|
try:
|
|
invoice_amount, zapped_event, sender, message, anon = await parse_zap_event_tags(zap_event,
|
|
self.keys,
|
|
self.dvm_config.NIP89.NAME,
|
|
self.client,
|
|
self.dvm_config)
|
|
user = await get_or_add_user(db=self.dvm_config.DB, npub=sender, client=self.client,
|
|
config=self.dvm_config)
|
|
|
|
if zapped_event is not None:
|
|
if zapped_event.kind() == EventDefinitions.KIND_FEEDBACK:
|
|
|
|
amount = 0
|
|
job_event = None
|
|
p_tag_str = ""
|
|
status = ""
|
|
for tag in zapped_event.tags().to_vec():
|
|
if tag.as_vec()[0] == 'amount':
|
|
amount = int(float(tag.as_vec()[1]) / 1000)
|
|
elif tag.as_vec()[0] == 'e':
|
|
job_event = await get_event_by_id(tag.as_vec()[1], client=self.client,
|
|
config=self.dvm_config)
|
|
if job_event is not None:
|
|
job_event, use_legacy_encryption = check_and_decrypt_tags(job_event, self.dvm_config)
|
|
if job_event is None:
|
|
return
|
|
else:
|
|
return
|
|
elif tag.as_vec()[0] == 'status':
|
|
status = tag.as_vec()[1]
|
|
print(status)
|
|
|
|
# if a reaction by us got zapped
|
|
print(status)
|
|
if job_event.kind() == EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT:
|
|
await send_job_status_reaction(job_event, "subscription-success", client=self.client,
|
|
dvm_config=self.dvm_config)
|
|
|
|
|
|
|
|
else:
|
|
task_supported, task = await check_task_is_supported(job_event, client=self.client,
|
|
config=self.dvm_config)
|
|
if job_event is not None and task_supported:
|
|
print("Zap received for NIP90 task: " + str(invoice_amount) + " Sats from " + str(
|
|
user.name + " (" + user.npub + ")"))
|
|
if amount <= invoice_amount:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Payment-request fulfilled...")
|
|
await send_job_status_reaction(job_event, "processing", client=self.client,
|
|
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
|
|
dvm_config=self.dvm_config)
|
|
indices = [i for i, x in enumerate(self.job_list) if
|
|
x.event == job_event]
|
|
index = -1
|
|
if len(indices) > 0:
|
|
index = indices[0]
|
|
if index > -1:
|
|
if self.job_list[index].is_processed:
|
|
self.job_list[index].is_paid = True
|
|
await check_and_return_event(self.job_list[index].result, job_event)
|
|
elif not (self.job_list[index]).is_processed:
|
|
# If payment-required appears before processing
|
|
self.job_list.pop(index)
|
|
print("Starting work...")
|
|
await do_work(job_event, invoice_amount)
|
|
else:
|
|
print("Job not in List, but starting work...")
|
|
await do_work(job_event, invoice_amount)
|
|
|
|
else:
|
|
await send_job_status_reaction(job_event, "payment-rejected",
|
|
False, invoice_amount, client=self.client,
|
|
dvm_config=self.dvm_config)
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Invoice was not paid sufficiently")
|
|
elif zapped_event.kind() == EventDefinitions.KIND_NIP88_SUBSCRIBE_EVENT:
|
|
print("new subscription, doing nothing")
|
|
|
|
elif zapped_event.kind() in EventDefinitions.ANY_RESULT:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] "
|
|
"Someone zapped the result of an exisiting Task. Nice")
|
|
elif not anon:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Note Zap received for DVM balance: " +
|
|
str(invoice_amount) + " Sats from " + str(user.name + " (" + user.npub + ")"))
|
|
# update_user_balance(self.dvm_config.DB, sender, invoice_amount, client=self.client,
|
|
# config=self.dvm_config)
|
|
|
|
# a regular note
|
|
elif not anon and dvm_config.NIP88 is None:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Profile Zap received for DVM balance: " +
|
|
str(invoice_amount) + " Sats from " + str(user.name + " (" + user.npub + ")"))
|
|
# update_user_balance(self.dvm_config.DB, sender, invoice_amount, client=self.client,
|
|
# config=self.dvm_config)
|
|
|
|
except Exception as e:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Error during content decryption: " + str(e))
|
|
|
|
async def check_event_has_not_unfinished_job_input(nevent, append, client, dvmconfig):
|
|
task_supported, task = await check_task_is_supported(nevent, client, config=dvmconfig)
|
|
if not task_supported:
|
|
return False
|
|
|
|
for tag in nevent.tags().to_vec():
|
|
if tag.as_vec()[0] == 'i':
|
|
if len(tag.as_vec()) < 3:
|
|
print("Job Event missing/malformed i tag, skipping..")
|
|
return False
|
|
else:
|
|
input = tag.as_vec()[1]
|
|
input_type = tag.as_vec()[2]
|
|
if input_type == "job":
|
|
evt = await get_referenced_event_by_id(event_id=input, client=client,
|
|
kinds=EventDefinitions.ANY_RESULT,
|
|
dvm_config=dvmconfig)
|
|
if evt is None:
|
|
if append:
|
|
job_ = RequiredJobToWatch(event=nevent, timestamp=Timestamp.now().as_secs())
|
|
self.jobs_on_hold_list.append(job_)
|
|
await send_job_status_reaction(nevent, "chain-scheduled", True, 0,
|
|
client=client, dvm_config=dvmconfig)
|
|
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
async def check_and_return_event(data, original_event: Event):
|
|
amount = 0
|
|
for x in self.job_list:
|
|
if x.event == original_event:
|
|
is_paid = x.is_paid
|
|
amount = x.amount
|
|
x.result = data
|
|
x.is_processed = True
|
|
if self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid:
|
|
await send_nostr_reply_event(data, original_event.as_json())
|
|
await send_job_status_reaction(original_event, "success", amount,
|
|
dvm_config=self.dvm_config
|
|
) # or payment-required, or both?
|
|
elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and not is_paid:
|
|
await send_job_status_reaction(original_event, "success", amount,
|
|
dvm_config=self.dvm_config
|
|
) # or payment-required, or both?
|
|
|
|
if self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and is_paid:
|
|
self.job_list.remove(x)
|
|
elif not self.dvm_config.SHOW_RESULT_BEFORE_PAYMENT and is_paid:
|
|
self.job_list.remove(x)
|
|
await send_nostr_reply_event(data, original_event.as_json())
|
|
break
|
|
|
|
task = await get_task(original_event, self.client, self.dvm_config)
|
|
for dvm in self.dvm_config.SUPPORTED_DVMS:
|
|
if task == dvm.TASK or dvm.TASK == "generic":
|
|
try:
|
|
post_processed = await dvm.post_process(data, original_event)
|
|
await send_nostr_reply_event(post_processed, original_event.as_json())
|
|
except Exception as e:
|
|
print(e)
|
|
# Zapping back by error in post-processing is a risk for the DVM because work has been done,
|
|
# but maybe something with parsing/uploading failed. Try to avoid errors here as good as possible
|
|
await send_job_status_reaction(original_event, "error",
|
|
content="Error in Post-processing: " + str(e),
|
|
dvm_config=self.dvm_config,
|
|
)
|
|
if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "":
|
|
user = await get_or_add_user(self.dvm_config.DB, original_event.author().to_hex(),
|
|
client=self.client, config=self.dvm_config)
|
|
print(user.lud16 + " " + str(amount))
|
|
bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats",
|
|
original_event, "",
|
|
self.keys, self.dvm_config.RELAY_LIST, zaptype="private")
|
|
if bolt11 is None:
|
|
print("Receiver has no Lightning address, can't zap back.")
|
|
return
|
|
try:
|
|
payment_hash = pay_bolt11_ln_bits(bolt11, self.dvm_config)
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
async def send_nostr_reply_event(content, original_event_as_str):
|
|
original_event = Event.from_json(original_event_as_str)
|
|
request_tag = Tag.parse(["request", original_event_as_str])
|
|
e_tag = Tag.parse(["e", original_event.id().to_hex()])
|
|
p_tag = Tag.parse(["p", original_event.author().to_hex()])
|
|
alt_tag = Tag.parse(["alt", "This is the result of a NIP90 DVM task with kind " + str(
|
|
original_event.kind().as_u16())])
|
|
status_tag = Tag.parse(["status", "success"])
|
|
reply_tags = [request_tag, e_tag, p_tag, alt_tag, status_tag]
|
|
|
|
relay_tag = None
|
|
for tag in original_event.tags().to_vec():
|
|
if tag.as_vec()[0] == "relays":
|
|
relay_tag = tag
|
|
if tag.as_vec()[0] == "client":
|
|
client = tag.as_vec()[1]
|
|
reply_tags.append(Tag.parse(["client", client]))
|
|
if relay_tag is not None:
|
|
reply_tags.append(relay_tag)
|
|
|
|
encrypted = False
|
|
is_legacy_encryption = False
|
|
encryption_tags = []
|
|
for tag in original_event.tags().to_vec():
|
|
|
|
|
|
if tag.as_vec()[0] == "encrypted":
|
|
encrypted = True
|
|
encrypted_tag = Tag.parse(["encrypted"])
|
|
encryption_tags.append(encrypted_tag)
|
|
#_, is_legacy_encryption = check_and_decrypt_tags(original_event, dvm_config)
|
|
|
|
|
|
for tag in original_event.tags().to_vec():
|
|
if tag.as_vec()[0] == "i":
|
|
if not encrypted:
|
|
reply_tags.append(tag)
|
|
elif tag.as_vec()[0] == "expiration":
|
|
reply_tags.append(tag)
|
|
|
|
if encrypted:
|
|
encryption_tags.append(p_tag)
|
|
encryption_tags.append(e_tag)
|
|
|
|
else:
|
|
reply_tags.append(p_tag)
|
|
|
|
if encrypted:
|
|
print(content)
|
|
if is_legacy_encryption:
|
|
content = nip04_encrypt(self.keys.secret_key(), PublicKey.parse(original_event.author().to_hex()),
|
|
content)
|
|
else:
|
|
|
|
content = nip44_encrypt(self.keys.secret_key(),
|
|
PublicKey.parse(original_event.author().to_hex()),
|
|
content, Nip44Version.V2)
|
|
|
|
reply_tags = encryption_tags
|
|
|
|
|
|
reply_event = EventBuilder(Kind(original_event.kind().as_u16() + 1000), str(content)).tags(reply_tags).sign_with_keys(
|
|
self.keys)
|
|
#print(reply_event)
|
|
# send_event(reply_event, client=self.client, dvm_config=self.dvm_config)
|
|
response_status = await send_event_outbox(reply_event, client=self.client, dvm_config=self.dvm_config)
|
|
if response_status is not None:
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print(bcolors.GREEN + "[" + self.dvm_config.NIP89.NAME + "] " + str(
|
|
original_event.kind().as_u16() + 1000) + " Job Response event sent: " + reply_event.as_json() + ". Success: " + str(
|
|
response_status.success) + " Failed: " + str(response_status.failed) + " EventID: "
|
|
+ response_status.id.to_hex() + " / " + response_status.id.to_bech32() + bcolors.ENDC)
|
|
|
|
elif self.dvm_config.LOGLEVEL.value >= LogLevel.INFO.value:
|
|
print(bcolors.GREEN + "[" + self.dvm_config.NIP89.NAME + "] " + str(
|
|
original_event.kind().as_u16() + 1000) + " Job Response event sent. Success: " + str(
|
|
response_status.success) + " Failed: " + str(response_status.failed) + " EventID: "
|
|
+ response_status.id.to_hex() + " / " + response_status.id.to_bech32() + bcolors.ENDC)
|
|
else:
|
|
print( bcolors.RED + "No success" + bcolors.ENDC)
|
|
|
|
async def send_job_status_reaction(original_event, status, is_paid=True, amount=0, client=None,
|
|
content=None,
|
|
dvm_config=None):
|
|
|
|
task = await get_task(original_event, client=client, dvm_config=dvm_config)
|
|
alt_description, reaction = build_status_reaction(status, task, amount, content, dvm_config)
|
|
|
|
e_tag = Tag.parse(["e", original_event.id().to_hex()])
|
|
p_tag = Tag.parse(["p", original_event.author().to_hex()])
|
|
alt_tag = Tag.parse(["alt", alt_description])
|
|
status_tag = Tag.parse(["status", status])
|
|
|
|
reply_tags = [e_tag, alt_tag, status_tag]
|
|
|
|
relay_tag = None
|
|
for tag in original_event.tags().to_vec():
|
|
if tag.as_vec()[0] == "relays":
|
|
relay_tag = tag
|
|
break
|
|
if relay_tag is not None:
|
|
reply_tags.append(relay_tag)
|
|
|
|
encryption_tags = []
|
|
|
|
encrypted = False
|
|
is_legacy_encryption = False
|
|
expiration_tag = None
|
|
for tag in original_event.tags().to_vec():
|
|
if tag.as_vec()[0] == "encrypted":
|
|
encrypted = True
|
|
encrypted_tag = Tag.parse(["encrypted"])
|
|
encryption_tags.append(encrypted_tag)
|
|
#_, is_legacy_encryption = check_and_decrypt_tags(original_event, dvm_config)
|
|
elif tag.as_vec()[0] == "expiration":
|
|
expiration_tag = tag
|
|
|
|
if encrypted:
|
|
encryption_tags.append(p_tag)
|
|
encryption_tags.append(e_tag)
|
|
|
|
else:
|
|
reply_tags.append(p_tag)
|
|
|
|
if status == "success" or status == "error": #
|
|
for x in self.job_list:
|
|
if x.event == original_event:
|
|
is_paid = x.is_paid
|
|
amount = x.amount
|
|
break
|
|
|
|
bolt11 = ""
|
|
payment_hash = ""
|
|
expires = original_event.created_at().as_secs() + (60 * 60 * 24)
|
|
if status == "payment-required" or (
|
|
status == "processing" and not is_paid):
|
|
if dvm_config.LNBITS_INVOICE_KEY != "" and dvm_config.PROVIDE_INVOICE:
|
|
try:
|
|
bolt11, payment_hash = create_bolt11_ln_bits(amount, dvm_config)
|
|
except Exception as e:
|
|
print(e)
|
|
try:
|
|
bolt11, payment_hash = create_bolt11_lud16(dvm_config.LN_ADDRESS,
|
|
amount)
|
|
except Exception as e:
|
|
print(e)
|
|
bolt11 = None
|
|
elif dvm_config.LN_ADDRESS != "" and dvm_config.PROVIDE_INVOICE:
|
|
try:
|
|
bolt11, payment_hash = create_bolt11_lud16(dvm_config.LN_ADDRESS, amount)
|
|
except Exception as e:
|
|
print(e)
|
|
bolt11 = None
|
|
else:
|
|
bolt11 = None
|
|
|
|
if not any(x.event == original_event for x in self.job_list):
|
|
self.job_list.append(
|
|
JobToWatch(event=original_event,
|
|
timestamp=original_event.created_at().as_secs(),
|
|
amount=amount,
|
|
is_paid=is_paid,
|
|
status=status, result="", is_processed=False, bolt11=bolt11,
|
|
payment_hash=payment_hash,
|
|
expires=expires))
|
|
# print(str(self.job_list))
|
|
if (status == "payment-required" or status == "payment-rejected" or (
|
|
status == "processing" and not is_paid)
|
|
or (status == "success" and not is_paid)):
|
|
|
|
if dvm_config.LNBITS_INVOICE_KEY != "" and bolt11 is not None:
|
|
amount_tag = Tag.parse(["amount", str(amount * 1000), bolt11])
|
|
else:
|
|
amount_tag = Tag.parse(["amount", str(amount * 1000)]) # to millisats
|
|
reply_tags.append(amount_tag)
|
|
|
|
if encrypted:
|
|
content_tag = Tag.parse(["content", reaction])
|
|
reply_tags.append(content_tag)
|
|
str_tags = []
|
|
for element in reply_tags:
|
|
str_tags.append(element.as_vec())
|
|
|
|
content = json.dumps(str_tags)
|
|
if is_legacy_encryption:
|
|
content = nip04_encrypt(self.keys.secret_key(), PublicKey.parse(original_event.author().to_hex()),
|
|
content)
|
|
else:
|
|
content = nip44_encrypt(self.keys.secret_key(),
|
|
PublicKey.parse(original_event.author().to_hex()),
|
|
content, version=Nip44Version.V2)
|
|
reply_tags = encryption_tags
|
|
|
|
else:
|
|
content = reaction
|
|
|
|
if expiration_tag is not None:
|
|
reply_tags.append(expiration_tag)
|
|
|
|
keys = Keys.parse(dvm_config.PRIVATE_KEY)
|
|
reaction_event = EventBuilder(EventDefinitions.KIND_FEEDBACK, str(content)).tags(reply_tags).sign_with_keys(keys)
|
|
# send_event(reaction_event, client=self.client, dvm_config=self.dvm_config)
|
|
response_status = await send_event_outbox(reaction_event, client=self.client, dvm_config=self.dvm_config)
|
|
if response_status is not None:
|
|
if self.dvm_config.LOGLEVEL.value >= LogLevel.DEBUG.value:
|
|
print(bcolors.YELLOW + "[" + self.dvm_config.NIP89.NAME + "] Sent Kind " + str(
|
|
EventDefinitions.KIND_FEEDBACK.as_u16()) + " Reaction: " + status + " " + reaction_event.as_json() + ". Success: " + str(
|
|
response_status.success) + " Failed: " + str(response_status.failed) + " EventID: "
|
|
+ response_status.id.to_hex() + " / " + response_status.id.to_bech32() + bcolors.ENDC)
|
|
|
|
elif self.dvm_config.LOGLEVEL.value >= LogLevel.INFO.value:
|
|
print(bcolors.YELLOW + "[" + self.dvm_config.NIP89.NAME + "] Sent Kind " + str(
|
|
EventDefinitions.KIND_FEEDBACK.as_u16()) + " Reaction: " + status + ". Success: " + str(
|
|
response_status.success) + " Failed: " + str(response_status.failed) + " EventID: "
|
|
+ response_status.id.to_hex() + " / " + response_status.id.to_bech32() + bcolors.ENDC)
|
|
else:
|
|
print(bcolors.RED + "No success" + bcolors.ENDC)
|
|
|
|
|
|
return reaction_event.as_json()
|
|
|
|
async def _read_stream(stream, cb):
|
|
while True:
|
|
line = await stream.readline()
|
|
if line:
|
|
cb(line)
|
|
else:
|
|
break
|
|
|
|
async def _stream_subprocess(cmd, stdout_cb, stderr_cb):
|
|
process = await asyncio.create_subprocess_exec(*cmd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE)
|
|
|
|
async def run_subprocess(python_bin, dvm_config, request_form, stdout_cb, stderr_cb):
|
|
print("Running subprocess, please wait..")
|
|
process = await asyncio.create_subprocess_exec(
|
|
python_bin, dvm_config.SCRIPT,
|
|
'--request', json.dumps(request_form),
|
|
'--identifier', dvm_config.IDENTIFIER,
|
|
'--output', 'output.txt',
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
|
|
await asyncio.gather(
|
|
_read_stream(process.stdout, stdout_cb),
|
|
_read_stream(process.stderr, stderr_cb)
|
|
)
|
|
return await process.wait()
|
|
|
|
# stdout, stderr = await process.communicate()
|
|
|
|
# retcode = process.returncode
|
|
|
|
# if retcode != 0:
|
|
# print(f"Error: {stderr.decode()}")
|
|
# else:
|
|
# print(f"Output: {stdout.decode()}")
|
|
|
|
# return retcode
|
|
|
|
async def do_work(job_event, amount):
|
|
if ((
|
|
EventDefinitions.KIND_NIP90_EXTRACT_TEXT.as_u16() <= job_event.kind().as_u16() <= EventDefinitions.KIND_NIP90_GENERIC.as_u16())
|
|
or job_event.kind().as_u16() == EventDefinitions.KIND_DM.as_u16()):
|
|
|
|
task = await get_task(job_event, client=self.client, dvm_config=self.dvm_config)
|
|
|
|
for dvm in self.dvm_config.SUPPORTED_DVMS:
|
|
result = ""
|
|
try:
|
|
if task == dvm.TASK or dvm.TASK == "generic":
|
|
|
|
request_form = await dvm.create_request_from_nostr_event(job_event, self.client,
|
|
self.dvm_config)
|
|
|
|
if dvm_config.USE_OWN_VENV:
|
|
python_location = "/bin/python"
|
|
if platform == "win32":
|
|
python_location = "/Scripts/python"
|
|
python_bin = (r'cache/venvs/' + os.path.basename(dvm_config.SCRIPT).split(".py")[0]
|
|
+ python_location)
|
|
# retcode = subprocess.call([python_bin, dvm_config.SCRIPT,
|
|
# '--request', json.dumps(request_form),
|
|
# '--identifier', dvm_config.IDENTIFIER,
|
|
# '--output', 'output.txt'])
|
|
await run_subprocess(python_bin, dvm_config, request_form,
|
|
lambda x: print("%s" % x.decode("utf-8").replace("\n", "")),
|
|
lambda x: print("STDERR: %s" % x.decode("utf-8")))
|
|
print("Finished processing, loading data..")
|
|
|
|
with open(os.path.abspath('output.txt'), encoding="utf-8") as f:
|
|
resultall = f.readlines()
|
|
for line in resultall:
|
|
if line != '\n':
|
|
result += line
|
|
os.remove(os.path.abspath('output.txt'))
|
|
assert not result.startswith("Error:")
|
|
print(result)
|
|
|
|
else: # Some components might have issues with running code in otuside venv.
|
|
# We install locally in these cases for now
|
|
result = await dvm.process(request_form)
|
|
try:
|
|
post_processed = await dvm.post_process(result, job_event)
|
|
await send_nostr_reply_event(post_processed, job_event.as_json())
|
|
except Exception as e:
|
|
print(bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str(
|
|
e) + bcolors.ENDC)
|
|
await send_job_status_reaction(job_event, "error", content=str(e),
|
|
dvm_config=self.dvm_config)
|
|
except Exception as e:
|
|
print(
|
|
bcolors.RED + "[" + self.dvm_config.NIP89.NAME + "] Error: " + str(e) + bcolors.ENDC)
|
|
|
|
# we could send the exception here to the user, but maybe that's not a good idea after all.
|
|
await send_job_status_reaction(job_event, "error", content=result,
|
|
dvm_config=self.dvm_config)
|
|
# Zapping back the user on error
|
|
if amount > 0 and self.dvm_config.LNBITS_ADMIN_KEY != "":
|
|
user = await get_or_add_user(self.dvm_config.DB, job_event.author().to_hex(),
|
|
client=self.client, config=self.dvm_config)
|
|
print(user.lud16 + " " + str(amount))
|
|
bolt11 = zaprequest(user.lud16, amount, "Couldn't finish job, returning sats", job_event,
|
|
PublicKey.parse(user.npub),
|
|
self.keys, self.dvm_config.RELAY_LIST, zaptype="private")
|
|
if bolt11 is None:
|
|
print("Receiver has no Lightning address, can't zap back.")
|
|
return
|
|
try:
|
|
payment_hash = pay_bolt11_ln_bits(bolt11, self.dvm_config)
|
|
except Exception as e:
|
|
print(e)
|
|
|
|
return
|
|
|
|
asyncio.create_task(self.client.handle_notifications(NotificationHandler()))
|
|
|
|
try:
|
|
|
|
while self.stop_thread is False:
|
|
for dvm in self.dvm_config.SUPPORTED_DVMS:
|
|
await dvm.schedule(self.dvm_config)
|
|
|
|
for job in self.job_list:
|
|
if job.bolt11 != "" and job.payment_hash != "" and not job.payment_hash is None and not job.is_paid:
|
|
ispaid = check_bolt11_ln_bits_is_paid(job.payment_hash, self.dvm_config)
|
|
if ispaid and job.is_paid is False:
|
|
print("is paid")
|
|
job.is_paid = True
|
|
amount = parse_amount_from_bolt11_invoice(job.bolt11)
|
|
|
|
job.is_paid = True
|
|
await send_job_status_reaction(job.event, "processing", True, 0,
|
|
content=self.dvm_config.CUSTOM_PROCESSING_MESSAGE,
|
|
client=self.client,
|
|
dvm_config=self.dvm_config)
|
|
print("[" + self.dvm_config.NIP89.NAME + "] doing work from joblist")
|
|
await do_work(job.event, amount)
|
|
elif ispaid is None: # invoice expired
|
|
self.job_list.remove(job)
|
|
|
|
if Timestamp.now().as_secs() > job.expires:
|
|
self.job_list.remove(job)
|
|
|
|
for job in self.jobs_on_hold_list:
|
|
if await check_event_has_not_unfinished_job_input(job.event, False, client=self.client,
|
|
dvmconfig=self.dvm_config):
|
|
await handle_nip90_job_event(nip90_event=job.event)
|
|
try:
|
|
self.jobs_on_hold_list.remove(job)
|
|
except:
|
|
print("[" + self.dvm_config.NIP89.NAME + "] Error removing Job on Hold from List after expiry")
|
|
|
|
if Timestamp.now().as_secs() > job.timestamp + 60 * 20: # remove jobs to look for after 20 minutes..
|
|
self.jobs_on_hold_list.remove(job)
|
|
|
|
await asyncio.sleep(1)
|
|
except BaseException:
|
|
print("end")
|
|
|
|
print("and now my watch has ended.")
|
|
|