1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274
|
// What is this module called?
module webhook;
// What does this module require to function?
import core.atomic : atomicOp;
import std.datetime;
import std.concurrency;
import std.json;
// What other modules that we have created do we need to import?
import arsd.cgi;
import config;
import onedrive;
import log;
import util;
class OneDriveWebhook {
private RequestServer server;
private string host;
private ushort port;
private Tid parentTid;
private bool started;
private ApplicationConfig appConfig;
private OneDriveApi oneDriveApiInstance;
string subscriptionId = "";
SysTime subscriptionExpiration, subscriptionLastErrorAt;
Duration subscriptionExpirationInterval, subscriptionRenewalInterval, subscriptionRetryInterval;
string notificationUrl = "";
private uint count;
this(Tid parentTid, ApplicationConfig appConfig) {
this.host = appConfig.getValueString("webhook_listening_host");
this.port = to!ushort(appConfig.getValueLong("webhook_listening_port"));
this.parentTid = parentTid;
this.appConfig = appConfig;
subscriptionExpiration = Clock.currTime(UTC());
subscriptionLastErrorAt = SysTime.fromUnixTime(0);
subscriptionExpirationInterval = dur!"seconds"(appConfig.getValueLong("webhook_expiration_interval"));
subscriptionRenewalInterval = dur!"seconds"(appConfig.getValueLong("webhook_renewal_interval"));
subscriptionRetryInterval = dur!"seconds"(appConfig.getValueLong("webhook_retry_interval"));
notificationUrl = appConfig.getValueString("webhook_public_url");
}
// The static serve() is necessary because spawn() does not like instance methods
void serve() {
if (this.started) {
return;
}
this.started = true;
this.count = 0;
server.listeningHost = this.host;
server.listeningPort = this.port;
spawn(&serveImpl, cast(shared) this);
addLogEntry("Started OneDrive API Webhook server");
// Subscriptions
oneDriveApiInstance = new OneDriveApi(this.appConfig);
oneDriveApiInstance.initialise();
createOrRenewSubscription();
}
void stop() {
if (!this.started)
return;
server.stop();
this.started = false;
addLogEntry("Stopped OneDrive API Webhook server");
object.destroy(server);
// Delete subscription if there exists any
try {
deleteSubscription();
} catch (OneDriveException e) {
logSubscriptionError(e);
}
// Release API instance back to the pool
oneDriveApiInstance.releaseCurlEngine();
object.destroy(oneDriveApiInstance);
oneDriveApiInstance = null;
}
private static void handle(shared OneDriveWebhook _this, Cgi cgi) {
if (debugHTTPSResponse) {
addLogEntry("Webhook request: " ~ to!string(cgi.requestMethod) ~ " " ~ to!string(cgi.requestUri));
if (!cgi.postBody.empty) {
addLogEntry("Webhook post body: " ~ to!string(cgi.postBody));
}
}
cgi.setResponseContentType("text/plain");
if ("validationToken" in cgi.get) {
// For validation requests, respond with the validation token passed in the query string
// https://docs.microsoft.com/en-us/onedrive/developer/rest-api/concepts/webhook-receiver-validation-request
cgi.write(cgi.get["validationToken"]);
addLogEntry("OneDrive API Webhook: handled validation request");
} else {
// Notifications don't include any information about the changes that triggered them.
// Put a refresh signal in the queue and let the main monitor loop process it.
// https://docs.microsoft.com/en-us/onedrive/developer/rest-api/concepts/using-webhooks
_this.count.atomicOp!"+="(1);
send(cast()_this.parentTid, to!ulong(_this.count));
cgi.write("OK");
addLogEntry("OneDrive API Webhook: sent refresh signal #" ~ to!string(_this.count));
}
}
private static void serveImpl(shared OneDriveWebhook _this) {
_this.server.serveEmbeddedHttp!(handle, OneDriveWebhook)(_this);
}
// Create a new subscription or renew the existing subscription
void createOrRenewSubscription() {
auto elapsed = Clock.currTime(UTC()) - subscriptionLastErrorAt;
if (elapsed < subscriptionRetryInterval) {
return;
}
try {
if (!hasValidSubscription()) {
createSubscription();
} else if (isSubscriptionUpForRenewal()) {
renewSubscription();
}
} catch (OneDriveException e) {
logSubscriptionError(e);
subscriptionLastErrorAt = Clock.currTime(UTC());
addLogEntry("Will retry creating or renewing subscription in " ~ to!string(subscriptionRetryInterval));
} catch (JSONException e) {
addLogEntry("ERROR: Unexpected JSON error when attempting to validate subscription: " ~ e.msg);
subscriptionLastErrorAt = Clock.currTime(UTC());
addLogEntry("Will retry creating or renewing subscription in " ~ to!string(subscriptionRetryInterval));
}
}
// Return the duration to next subscriptionExpiration check
Duration getNextExpirationCheckDuration() {
SysTime now = Clock.currTime(UTC());
if (hasValidSubscription()) {
Duration elapsed = Clock.currTime(UTC()) - subscriptionLastErrorAt;
// Check if we are waiting for the next retry
if (elapsed < subscriptionRetryInterval)
return subscriptionRetryInterval - elapsed;
else
return subscriptionExpiration - now - subscriptionRenewalInterval;
}
else
return subscriptionRetryInterval;
}
private bool hasValidSubscription() {
return !subscriptionId.empty && subscriptionExpiration > Clock.currTime(UTC());
}
private bool isSubscriptionUpForRenewal() {
return subscriptionExpiration < Clock.currTime(UTC()) + subscriptionRenewalInterval;
}
private void createSubscription() {
addLogEntry("Initialising webhook subscription for updates ...");
auto expirationDateTime = Clock.currTime(UTC()) + subscriptionExpirationInterval;
try {
JSONValue response = oneDriveApiInstance.createSubscription(notificationUrl, expirationDateTime);
// Save important subscription metadata including id and expiration
subscriptionId = response["id"].str;
subscriptionExpiration = SysTime.fromISOExtString(response["expirationDateTime"].str);
addLogEntry("Created new subscription " ~ subscriptionId ~ " with expiration: " ~ to!string(subscriptionExpiration.toISOExtString()));
} catch (OneDriveException e) {
if (e.httpStatusCode == 409) {
// Take over an existing subscription on HTTP 409.
//
// Sample 409 error:
// {
// "error": {
// "code": "ObjectIdentifierInUse",
// "innerError": {
// "client-request-id": "615af209-467a-4ab7-8eff-27c1d1efbc2d",
// "date": "2023-09-26T09:27:45",
// "request-id": "615af209-467a-4ab7-8eff-27c1d1efbc2d"
// },
// "message": "Subscription Id c0bba80e-57a3-43a7-bac2-e6f525a76e7c already exists for the requested combination"
// }
// }
// Make sure the error code is "ObjectIdentifierInUse"
try {
if (e.error["error"]["code"].str != "ObjectIdentifierInUse") {
throw e;
}
} catch (JSONException jsonEx) {
throw e;
}
// Extract the existing subscription id from the error message
import std.regex;
auto idReg = ctRegex!(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", "i");
auto m = matchFirst(e.error["error"]["message"].str, idReg);
if (!m) {
throw e;
}
// Save the subscription id and renew it immediately since we don't know the expiration timestamp
subscriptionId = m[0];
addLogEntry("Found existing webhook subscription " ~ subscriptionId);
renewSubscription();
} else {
throw e;
}
}
}
private void renewSubscription() {
addLogEntry("Renewing webhook subscription for updates ...");
auto expirationDateTime = Clock.currTime(UTC()) + subscriptionExpirationInterval;
try {
JSONValue response = oneDriveApiInstance.renewSubscription(subscriptionId, expirationDateTime);
// Update subscription expiration from the response
subscriptionExpiration = SysTime.fromISOExtString(response["expirationDateTime"].str);
addLogEntry("Renewed webhook subscription " ~ subscriptionId ~ " with expiration: " ~ to!string(subscriptionExpiration.toISOExtString()));
} catch (OneDriveException e) {
if (e.httpStatusCode == 404) {
addLogEntry("The subscription is not found on the server. Recreating subscription ...");
subscriptionId = null;
subscriptionExpiration = Clock.currTime(UTC());
createSubscription();
} else {
throw e;
}
}
}
private void deleteSubscription() {
if (!hasValidSubscription()) {
return;
}
oneDriveApiInstance.deleteSubscription(subscriptionId);
addLogEntry("Deleted subscription");
}
private void logSubscriptionError(OneDriveException e) {
string errorMsg;
try {
// Attempt to extract the specific error message from the JSON if possible
if (e.error.type == JSONType.object &&
"error" in e.error &&
e.error["error"].type == JSONType.object &&
"message" in e.error["error"]) {
errorMsg = e.error["error"]["message"].str;
} else {
throw new Exception("Invalid error structure");
}
} catch (Exception ex) {
// Fallback to the message stored in the exception if the JSON is malformed or not structured as expected
errorMsg = e.msg;
}
// Log a message to the GUI only
addLogEntry("ERROR: An issue has occurred with webhook subscriptions: " ~ errorMsg, ["notify"]);
// Use the standard OneDrive API logging method
displayOneDriveErrorMessage(errorMsg, getFunctionName!({}));
}
}
|