Add revenue split to affiliate codes v2

apps/labrinth/.env.docker-compose

@@ -146,3 +146,5 @@ GOTENBERG_URL=http://labrinth-gotenberg:13000
 GOTENBERG_CALLBACK_BASE=http://host.docker.internal:8000/_internal/gotenberg
 
 ARCHON_URL=none
+
+DEFAULT_AFFILIATE_REVENUE_SPLIT=0.1

apps/labrinth/.env.local

@@ -147,3 +147,5 @@ GOTENBERG_URL=http://localhost:13000
 GOTENBERG_CALLBACK_BASE=http://host.docker.internal:8000/_internal/gotenberg
 
 ARCHON_URL=none
+
+DEFAULT_AFFILIATE_REVENUE_SPLIT=0.1

apps/labrinth/migrations/20251024182919_subscription_affiliations.sql

@@ -0,0 +1,19 @@
+CREATE TABLE users_subscriptions_affiliations (
+	subscription_id BIGINT      NOT NULL REFERENCES users_subscriptions(id),
+	affiliate_code  BIGINT      NOT NULL REFERENCES affiliate_codes(id),
+	created_at      TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
+	deactivated_at  TIMESTAMPTZ,
+	UNIQUE (subscription_id)
+);
+
+CREATE TABLE users_subscriptions_affiliations_payouts(
+	id              BIGSERIAL          PRIMARY KEY,
+	charge_id       BIGINT    NOT NULL REFERENCES charges(id),
+	subscription_id BIGINT    NOT NULL REFERENCES users_subscriptions(id),
+	affiliate_code  BIGINT    NOT NULL REFERENCES affiliate_codes(id),
+	payout_value_id BIGSERIAL NOT NULL REFERENCES payouts_values(id),
+	UNIQUE (charge_id)
+);
+
+ALTER TABLE payouts_values
+ADD COLUMN affiliate_code_source BIGINT;

apps/labrinth/src/background_task.rs

@@ -3,7 +3,8 @@ use crate::queue::billing::{index_billing, index_subscriptions};
 use crate::queue::email::EmailQueue;
 use crate::queue::payouts::{
     PayoutsQueue, index_payouts_notifications,
-    insert_bank_balances_and_webhook, process_payout,
+    insert_bank_balances_and_webhook, process_affiliate_payouts,
+    process_payout,
 };
 use crate::search::indexing::index_projects;
 use crate::util::anrok;
@@ -179,12 +180,17 @@ pub async fn payouts(
     info!("Started running payouts");
     let result = process_payout(&pool, &clickhouse).await;
     if let Err(e) = result {
-        warn!("Payouts run failed: {:?}", e);
+        warn!("Payouts run failed: {e:#?}");
     }
 
     let result = index_payouts_notifications(&pool, &redis_pool).await;
     if let Err(e) = result {
-        warn!("Payouts notifications indexing failed: {:?}", e);
+        warn!("Payouts notifications indexing failed: {e:#?}");
+    }
+
+    let result = process_affiliate_payouts(&pool).await;
+    if let Err(e) = result {
+        warn!("Affiliate payouts run failed: {e:#?}");
     }
 
     info!("Done running payouts");

apps/labrinth/src/database/models/mod.rs

@@ -35,6 +35,7 @@ pub mod user_subscription_item;
 pub mod users_compliance;
 pub mod users_notifications_preferences_item;
 pub mod users_redeemals;
+pub mod users_subscriptions_affiliations;
 pub mod users_subscriptions_credits;
 pub mod version_item;
 

apps/labrinth/src/database/models/users_subscriptions_affiliations.rs

@@ -0,0 +1,86 @@
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+use crate::database::models::{
+    DBAffiliateCodeId, DBChargeId, DBUserSubscriptionId,
+};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DBUsersSubscriptionsAffiliations {
+    pub subscription_id: DBUserSubscriptionId,
+    pub affiliate_code: DBAffiliateCodeId,
+    pub deactivated_at: Option<DateTime<Utc>>,
+}
+
+impl DBUsersSubscriptionsAffiliations {
+    pub async fn insert<'a, E>(&self, exec: E) -> sqlx::Result<()>
+    where
+        E: sqlx::PgExecutor<'a>,
+    {
+        sqlx::query_scalar!(
+            "
+            INSERT INTO users_subscriptions_affiliations
+                (subscription_id, affiliate_code, deactivated_at)
+            VALUES ($1, $2, $3)
+            ",
+            self.subscription_id.0,
+            self.affiliate_code.0,
+            self.deactivated_at,
+        )
+        .fetch_one(exec)
+        .await?;
+        Ok(())
+    }
+
+    pub async fn deactivate<'a, E>(
+        subscription_id: DBUserSubscriptionId,
+        exec: E,
+    ) -> sqlx::Result<()>
+    where
+        E: sqlx::PgExecutor<'a>,
+    {
+        sqlx::query!(
+            "UPDATE users_subscriptions_affiliations
+            SET deactivated_at = NOW()
+            WHERE subscription_id = $1",
+            subscription_id.0,
+        )
+        .execute(exec)
+        .await?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DBUsersSubscriptionsAffiliationsPayouts {
+    pub id: i64,
+    pub charge_id: DBChargeId,
+    pub subscription_id: DBUserSubscriptionId,
+    pub affiliate_code: DBAffiliateCodeId,
+    pub payout_value_id: i64,
+}
+
+impl DBUsersSubscriptionsAffiliationsPayouts {
+    pub async fn insert<'a, E>(&mut self, exec: E) -> sqlx::Result<()>
+    where
+        E: sqlx::PgExecutor<'a>,
+    {
+        let id = sqlx::query_scalar!(
+            "
+            INSERT INTO users_subscriptions_affiliations_payouts
+                (charge_id, subscription_id, affiliate_code, payout_value_id)
+            VALUES ($1, $2, $3, $4)
+            RETURNING id
+            ",
+            self.charge_id.0,
+            self.subscription_id.0,
+            self.affiliate_code.0,
+            self.payout_value_id,
+        )
+        .fetch_one(exec)
+        .await?;
+
+        self.id = id;
+        Ok(())
+    }
+}

apps/labrinth/src/lib.rs

@@ -526,5 +526,7 @@ pub fn check_env_vars() -> bool {
 
     failed |= check_var::<String>("ARCHON_URL");
 
+    failed |= check_var::<String>("DEFAULT_AFFILIATE_REVENUE_SPLIT");
+
     failed
 }

apps/labrinth/src/queue/payouts.rs

@@ -26,6 +26,9 @@ use std::collections::HashMap;
 use tokio::sync::RwLock;
 use tracing::{error, info};
 
+mod affiliate;
+pub use affiliate::process_affiliate_payouts;
+
 pub struct PayoutsQueue {
     credential: RwLock<Option<PayPalCredentials>>,
     payout_options: RwLock<Option<PayoutMethods>>,