feat!: due date reminder notifications

.tmuxinator.yml

@@ -10,6 +10,8 @@ windows:
         - yarn:
           - cd frontend
           - yarn start
+        - worker:
+          - go run cmd/taskcafe/main.go worker
   - web/editor:
       root: ./frontend
       panes:

docker-compose.dev.yml

@@ -19,17 +19,11 @@ services:
     ports:
       - 1025:1025
       - 8025:8025
-  broker:
+  redis:
-    image: rabbitmq:3-management
+    image: redis:6.2
     restart: always
     ports:
-      - 8060:15672
+      - 6379:6379
-      - 5672:5672
-  result_store:
-    image: memcached:1.6-alpine
-    restart: always
-    ports:
-      - 11211:11211
 
 volumes:
   taskcafe-postgres:

frontend/src/App/TopNavbar/index.tsx

@@ -73,7 +73,9 @@ const LoggedInNavbar: React.FC<GlobalTopNavbarProps> = ({
   });
   const { showPopup, hidePopup } = usePopup();
   const { setUser } = useCurrentUser();
-  const { data: unreadData } = useHasUnreadNotificationsQuery({ pollInterval: polling.UNREAD_NOTIFICATIONS });
+  const { data: unreadData, refetch: refetchHasUnread } = useHasUnreadNotificationsQuery({
+    pollInterval: polling.UNREAD_NOTIFICATIONS,
+  });
   const history = useHistory();
   const onLogout = () => {
     fetch('/auth/logout', {
@@ -118,9 +120,11 @@ const LoggedInNavbar: React.FC<GlobalTopNavbarProps> = ({
 
   // TODO: rewrite popup to contain subscription and notification fetch
   const onNotificationClick = ($target: React.RefObject<HTMLElement>) => {
-    if (data) {
+    showPopup($target, <NotificationPopup onToggleRead={() => refetchHasUnread()} />, {
-      showPopup($target, <NotificationPopup />, { width: 605, borders: false, diamondColor: theme.colors.primary });
+      width: 605,
-    }
+      borders: false,
+      diamondColor: theme.colors.primary,
+    });
   };
 
   // TODO: readd permision check

frontend/src/shared/components/DueDateManager/index.tsx

@@ -446,10 +446,8 @@ const DueDateManager: React.FC<DueDateManagerProps> = ({ task, onDueDateChange,
         </LeftWrapper>
         <RightWrapper>
           <ActionIcon
-            // disabled={notifications.length === 3}
+            disabled={notifications.length === 3}
-            disabled
             onClick={() => {
-              /*
               setNotifications((prev) => [
                 ...prev,
                 {
@@ -459,7 +457,6 @@ const DueDateManager: React.FC<DueDateManagerProps> = ({ task, onDueDateChange,
                   period: 10,
                 },
               ]);
-               */
             }}
           >
             <Bell width={16} height={16} />

frontend/src/shared/components/NotifcationPopup/index.tsx

@@ -1,9 +1,10 @@
-import React, { useState } from 'react';
+import React, { useRef, useState } from 'react';
 import styled, { css } from 'styled-components';
 import TimeAgo from 'react-timeago';
 import { Link } from 'react-router-dom';
 import { mixin } from 'shared/utils/styles';
 import {
+  useNotificationMarkAllReadMutation,
   useNotificationsQuery,
   NotificationFilter,
   ActionType,
@@ -13,10 +14,24 @@ import {
 import dayjs from 'dayjs';
 
 import { Popup, usePopup } from 'shared/components/PopupMenu';
-import { CheckCircleOutline, Circle, CircleSolid, UserCircle } from 'shared/icons';
+import { Bell, CheckCircleOutline, Circle, Ellipsis, UserCircle } from 'shared/icons';
 import produce from 'immer';
 import { useLocalStorage } from 'shared/hooks/useStateWithLocalStorage';
 import localStorage from 'shared/utils/localStorage';
+import useOnOutsideClick from 'shared/hooks/onOutsideClick';
+
+function getFilterMessage(filter: NotificationFilter) {
+  switch (filter) {
+    case NotificationFilter.Unread:
+      return 'no unread';
+    case NotificationFilter.Assigned:
+      return 'no assigned';
+    case NotificationFilter.Mentioned:
+      return 'no mentioned';
+    default:
+      return 'no';
+  }
+}
 
 const ItemWrapper = styled.div`
   cursor: pointer;
@@ -98,6 +113,17 @@ const NotificationHeaderTitle = styled.span`
   color: ${(props) => props.theme.colors.text.secondary};
 `;
 
+const EmptyMessage = styled.div`
+  display: flex;
+  justify-content: center;
+  align-items: center;
+  font-size: 14px;
+  height: 448px;
+`;
+
+const EmptyMessageLabel = styled.span`
+  margin-bottom: 80px;
+`;
 const Notifications = styled.div`
   border-right: 1px solid rgba(0, 0, 0, 0.1);
   border-left: 1px solid rgba(0, 0, 0, 0.1);
@@ -180,7 +206,6 @@ const NotificationTab = styled.div<{ active: boolean }>`
 
 const NotificationLink = styled(Link)`
   display: flex;
-  align-items: center;
   text-decoration: none;
   padding: 16px 8px;
   width: 100%;
@@ -213,8 +238,8 @@ const NotificationButton = styled.div`
   }
 `;
 
-const NotificationWrapper = styled.li`
+const NotificationWrapper = styled.li<{ read: boolean }>`
-  min-height: 112px;
+  min-height: 80px;
   display: flex;
   font-size: 14px;
   transition: background-color 0.1s ease-in-out;
@@ -231,20 +256,28 @@ const NotificationWrapper = styled.li`
   &:hover ${NotificationControls} {
     visibility: visible;
   }
+  ${(props) =>
+    !props.read &&
+    css`
+      background: ${(props) => mixin.rgba(props.theme.colors.primary, 0.5)};
+      &:hover {
+        background: ${(props) => mixin.rgba(props.theme.colors.primary, 0.6)};
+      }
+    `}
 `;
 
 const NotificationContentFooter = styled.div`
-  margin-top: 8px;
+  margin-top: 10px;
   display: flex;
   align-items: center;
   color: ${(props) => props.theme.colors.text.primary};
 `;
 
 const NotificationCausedBy = styled.div`
-  height: 60px;
+  height: 48px;
-  width: 60px;
+  width: 48px;
-  min-height: 60px;
+  min-height: 48px;
-  min-width: 60px;
+  min-width: 48px;
 `;
 const NotificationCausedByInitials = styled.div`
   position: relative;
@@ -292,7 +325,6 @@ const NotificationContentHeader = styled.div`
 `;
 
 const NotificationBody = styled.div`
-  margin-top: 8px;
   display: flex;
   align-items: center;
   color: #fff;
@@ -328,17 +360,39 @@ const Notification: React.FC<NotificationProps> = ({ causedBy, createdAt, data,
   let link = '#';
   switch (actionType) {
     case ActionType.TaskAssigned:
-      prefix.push(<UserCircle width={14} height={16} />);
+      prefix.push(<UserCircle key="profile" width={14} height={16} />);
-      prefix.push(<NotificationPrefix>Assigned </NotificationPrefix>);
+      prefix.push(
-      prefix.push(<span>you to the task "{dataMap.get('TaskName')}"</span>);
+        <NotificationPrefix key="prefix">
+          <span style={{ fontWeight: 'bold' }}>{causedBy ? causedBy.fullname : 'Removed user'}</span>
+        </NotificationPrefix>,
+      );
+      prefix.push(<span key="content">assigned you to the task "{dataMap.get('TaskName')}"</span>);
       link = `/p/${dataMap.get('ProjectID')}/board/c/${dataMap.get('TaskID')}`;
       break;
+    case ActionType.DueDateReminder:
+      prefix.push(<Bell key="profile" width={14} height={16} />);
+      prefix.push(<NotificationPrefix key="prefix">{dataMap.get('TaskName')}</NotificationPrefix>);
+      const now = dayjs();
+      if (dayjs(dataMap.get('DueDate')).isBefore(dayjs())) {
+        prefix.push(
+          <span key="content">is due {dayjs.duration(now.diff(dayjs(dataMap.get('DueAt')))).humanize(true)}</span>,
+        );
+      } else {
+        prefix.push(
+          <span key="content">
+            has passed the due date {dayjs.duration(dayjs(dataMap.get('DueAt')).diff(now)).humanize(true)}
+          </span>,
+        );
+      }
+      link = `/p/${dataMap.get('ProjectID')}/board/c/${dataMap.get('TaskID')}`;
+      break;
+
     default:
       throw new Error('unknown action type');
   }
 
   return (
-    <NotificationWrapper>
+    <NotificationWrapper read={read}>
       <NotificationLink to={link} onClick={hidePopup}>
         <NotificationCausedBy>
           <NotificationCausedByInitials>
@@ -351,10 +405,6 @@ const Notification: React.FC<NotificationProps> = ({ causedBy, createdAt, data,
           </NotificationCausedByInitials>
         </NotificationCausedBy>
         <NotificationContent>
-          <NotificationContentHeader>
-            {causedBy ? causedBy.fullname : 'Removed user'}
-            {!read && <CircleSolid width={10} height={10} />}
-          </NotificationContentHeader>
           <NotificationBody>{prefix}</NotificationBody>
           <NotificationContentFooter>
             <span>{dayjs.duration(dayjs(createdAt).diff(dayjs())).humanize(true)}</span>
@@ -404,7 +454,59 @@ type NotificationEntry = {
     createdAt: string;
   };
 };
-const NotificationPopup: React.FC = ({ children }) => {
+type NotificationPopupProps = {
+  onToggleRead: () => void;
+};
+
+const NotificationHeaderMenu = styled.div`
+  position: absolute;
+  right: 16px;
+  top: 16px;
+`;
+
+const NotificationHeaderMenuIcon = styled.div`
+  cursor: pointer;
+  display: flex;
+  align-items: center;
+  justify-content: center;
+  position: relative;
+  svg {
+    fill: #fff;
+    stroke: #fff;
+  }
+`;
+
+const NotificationHeaderMenuContent = styled.div<{ show: boolean }>`
+  min-width: 130px;
+  position: absolute;
+  top: 16px;
+  background: #fff;
+  border-radius: 6px;
+  height: 50px;
+  visibility: ${(props) => (props.show ? 'visible' : 'hidden')};
+  border: 1px solid rgba(0, 0, 0, 0.1);
+  border-color: #414561;
+  background: #262c49;
+  padding: 6px;
+  display: flex;
+  flex-direction: column;
+`;
+
+const NotificationHeaderMenuButton = styled.div`
+  position: relative;
+  padding-left: 4px;
+  padding-right: 4px;
+  padding-top: 0.5rem;
+  padding-bottom: 0.5rem;
+  cursor: pointer;
+  display: flex;
+  align-items: center;
+  font-size: 14px;
+  &:hover {
+    background: ${(props) => props.theme.colors.primary};
+  }
+`;
+const NotificationPopup: React.FC<NotificationPopupProps> = ({ onToggleRead }) => {
   const [filter, setFilter] = useLocalStorage<NotificationFilter>(
     localStorage.NOTIFICATIONS_FILTER,
     NotificationFilter.Unread,
@@ -425,10 +527,12 @@ const NotificationPopup: React.FC = ({ children }) => {
           }
         });
       });
+      onToggleRead();
     },
   });
-  const { data: nData, fetchMore } = useNotificationsQuery({
+  const { fetchMore } = useNotificationsQuery({
-    variables: { limit: 5, filter },
+    variables: { limit: 8, filter },
+    fetchPolicy: 'network-only',
     onCompleted: (d) => {
       setData((prev) => ({
         hasNextPage: d.notified.pageInfo.hasNextPage,
@@ -437,7 +541,7 @@ const NotificationPopup: React.FC = ({ children }) => {
       }));
     },
   });
-  const { data: sData, loading } = useNotificationAddedSubscription({
+  useNotificationAddedSubscription({
     onSubscriptionData: (d) => {
       setData((n) => {
         if (d.subscriptionData.data) {
@@ -450,12 +554,40 @@ const NotificationPopup: React.FC = ({ children }) => {
       });
     },
   });
+  const [toggleAllRead] = useNotificationMarkAllReadMutation();
 
+  const [showHeaderMenu, setShowHeaderMenu] = useState(false);
+
+  const $menuContent = useRef<HTMLDivElement>(null);
+  useOnOutsideClick($menuContent, true, () => setShowHeaderMenu(false), null);
   return (
     <Popup title={null} tab={0} borders={false} padding={false}>
       <PopupContent>
         <NotificationHeader>
           <NotificationHeaderTitle>Notifications</NotificationHeaderTitle>
+          <NotificationHeaderMenu>
+            <NotificationHeaderMenuIcon onClick={() => setShowHeaderMenu(true)}>
+              <Ellipsis size={18} color="#fff" vertical={false} />
+              <NotificationHeaderMenuContent ref={$menuContent} show={showHeaderMenu}>
+                <NotificationHeaderMenuButton
+                  onClick={(e) => {
+                    e.stopPropagation();
+                    setShowHeaderMenu(() => false);
+                    toggleAllRead().then(() => {
+                      setData((prev) =>
+                        produce(prev, (draftData) => {
+                          draftData.nodes = draftData.nodes.map((node) => ({ ...node, read: true }));
+                        }),
+                      );
+                      onToggleRead();
+                    });
+                  }}
+                >
+                  Mark all as read
+                </NotificationHeaderMenuButton>
+              </NotificationHeaderMenuContent>
+            </NotificationHeaderMenuIcon>
+          </NotificationHeaderMenu>
         </NotificationHeader>
         <NotificationTabs>
           {tabs.map((tab) => (
@@ -473,65 +605,73 @@ const NotificationPopup: React.FC = ({ children }) => {
             </NotificationTab>
           ))}
         </NotificationTabs>
-        <Notifications
+        {data.nodes.length !== 0 ? (
-          onScroll={({ currentTarget }) => {
+          <Notifications
-            if (currentTarget.scrollTop + currentTarget.clientHeight >= currentTarget.scrollHeight) {
+            onScroll={({ currentTarget }) => {
-              if (data.hasNextPage) {
+              if (Math.ceil(currentTarget.scrollTop + currentTarget.clientHeight) >= currentTarget.scrollHeight) {
-                console.log(`fetching more = ${data.cursor} - ${data.hasNextPage}`);
+                if (data.hasNextPage) {
-                fetchMore({
+                  console.log(`fetching more = ${data.cursor} - ${data.hasNextPage}`);
-                  variables: {
+                  fetchMore({
-                    limit: 5,
+                    variables: {
-                    filter,
+                      limit: 8,
-                    cursor: data.cursor,
+                      filter,
-                  },
+                      cursor: data.cursor,
-                  updateQuery: (prev, { fetchMoreResult }) => {
-                    if (!fetchMoreResult) return prev;
-                    setData((d) => ({
-                      cursor: fetchMoreResult.notified.pageInfo.endCursor ?? '',
-                      hasNextPage: fetchMoreResult.notified.pageInfo.hasNextPage,
-                      nodes: [...d.nodes, ...fetchMoreResult.notified.notified],
-                    }));
-                    return {
-                      ...prev,
-                      notified: {
-                        ...prev.notified,
-                        pageInfo: {
-                          ...fetchMoreResult.notified.pageInfo,
-                        },
-                        notified: [...prev.notified.notified, ...fetchMoreResult.notified.notified],
-                      },
-                    };
-                  },
-                });
-              }
-            }
-          }}
-        >
-          {data.nodes.map((n) => (
-            <Notification
-              key={n.id}
-              read={n.read}
-              actionType={n.notification.actionType}
-              data={n.notification.data}
-              createdAt={n.notification.createdAt}
-              causedBy={n.notification.causedBy}
-              onToggleRead={() =>
-                toggleRead({
-                  variables: { notifiedID: n.id },
-                  optimisticResponse: {
-                    __typename: 'Mutation',
-                    notificationToggleRead: {
-                      __typename: 'Notified',
-                      id: n.id,
-                      read: !n.read,
-                      readAt: new Date().toUTCString(),
                     },
-                  },
+                    updateQuery: (prev, { fetchMoreResult }) => {
-                })
+                      if (!fetchMoreResult) return prev;
+                      setData((d) => ({
+                        cursor: fetchMoreResult.notified.pageInfo.endCursor ?? '',
+                        hasNextPage: fetchMoreResult.notified.pageInfo.hasNextPage,
+                        nodes: [...d.nodes, ...fetchMoreResult.notified.notified],
+                      }));
+                      return {
+                        ...prev,
+                        notified: {
+                          ...prev.notified,
+                          pageInfo: {
+                            ...fetchMoreResult.notified.pageInfo,
+                          },
+                          notified: [...prev.notified.notified, ...fetchMoreResult.notified.notified],
+                        },
+                      };
+                    },
+                  });
+                }
               }
-            />
+            }}
-          ))}
+          >
-        </Notifications>
+            {data.nodes.map((n) => (
+              <Notification
+                key={n.id}
+                read={n.read}
+                actionType={n.notification.actionType}
+                data={n.notification.data}
+                createdAt={n.notification.createdAt}
+                causedBy={n.notification.causedBy}
+                onToggleRead={() =>
+                  toggleRead({
+                    variables: { notifiedID: n.id },
+                    optimisticResponse: {
+                      __typename: 'Mutation',
+                      notificationToggleRead: {
+                        __typename: 'Notified',
+                        id: n.id,
+                        read: !n.read,
+                        readAt: new Date().toUTCString(),
+                      },
+                    },
+                  }).then(() => {
+                    onToggleRead();
+                  })
+                }
+              />
+            ))}
+          </Notifications>
+        ) : (
+          <EmptyMessage>
+            <EmptyMessageLabel>You have {getFilterMessage(filter)} notifications</EmptyMessageLabel>
+          </EmptyMessage>
+        )}
       </PopupContent>
     </Popup>
   );

frontend/src/shared/generated/graphql.tsx

@@ -34,6 +34,7 @@ export enum ActionType {
   DueDateAdded = 'DUE_DATE_ADDED',
   DueDateRemoved = 'DUE_DATE_REMOVED',
   DueDateChanged = 'DUE_DATE_CHANGED',
+  DueDateReminder = 'DUE_DATE_REMINDER',
   TaskAssigned = 'TASK_ASSIGNED',
   TaskMoved = 'TASK_MOVED',
   TaskArchived = 'TASK_ARCHIVED',
@@ -456,6 +457,7 @@ export type Mutation = {
   duplicateTaskGroup: DuplicateTaskGroupPayload;
   inviteProjectMembers: InviteProjectMembersPayload;
   logoutUser: Scalars['Boolean'];
+  notificationMarkAllRead: NotificationMarkAllAsReadResult;
   notificationToggleRead: Notified;
   removeTaskLabel: Task;
   setTaskChecklistItemComplete: TaskChecklistItem;
@@ -899,6 +901,11 @@ export enum NotificationFilter {
   Mentioned = 'MENTIONED'
 }
 
+export type NotificationMarkAllAsReadResult = {
+  __typename?: 'NotificationMarkAllAsReadResult';
+  success: Scalars['Boolean'];
+};
+
 export type NotificationToggleReadInput = {
   notifiedID: Scalars['UUID'];
 };
@@ -1929,6 +1936,17 @@ export type NotificationsQuery = (
   ) }
 );
 
+export type NotificationMarkAllReadMutationVariables = Exact<{ [key: string]: never; }>;
+
+
+export type NotificationMarkAllReadMutation = (
+  { __typename?: 'Mutation' }
+  & { notificationMarkAllRead: (
+    { __typename?: 'NotificationMarkAllAsReadResult' }
+    & Pick<NotificationMarkAllAsReadResult, 'success'>
+  ) }
+);
+
 export type NotificationAddedSubscriptionVariables = Exact<{ [key: string]: never; }>;
 
 
@@ -3891,6 +3909,38 @@ export function useNotificationsLazyQuery(baseOptions?: Apollo.LazyQueryHookOpti
 export type NotificationsQueryHookResult = ReturnType<typeof useNotificationsQuery>;
 export type NotificationsLazyQueryHookResult = ReturnType<typeof useNotificationsLazyQuery>;
 export type NotificationsQueryResult = Apollo.QueryResult<NotificationsQuery, NotificationsQueryVariables>;
+export const NotificationMarkAllReadDocument = gql`
+    mutation notificationMarkAllRead {
+  notificationMarkAllRead {
+    success
+  }
+}
+    `;
+export type NotificationMarkAllReadMutationFn = Apollo.MutationFunction<NotificationMarkAllReadMutation, NotificationMarkAllReadMutationVariables>;
+
+/**
+ * __useNotificationMarkAllReadMutation__
+ *
+ * To run a mutation, you first call `useNotificationMarkAllReadMutation` within a React component and pass it any options that fit your needs.
+ * When your component renders, `useNotificationMarkAllReadMutation` returns a tuple that includes:
+ * - A mutate function that you can call at any time to execute the mutation
+ * - An object with fields that represent the current status of the mutation's execution
+ *
+ * @param baseOptions options that will be passed into the mutation, supported options are listed on: https://www.apollographql.com/docs/react/api/react-hooks/#options-2;
+ *
+ * @example
+ * const [notificationMarkAllReadMutation, { data, loading, error }] = useNotificationMarkAllReadMutation({
+ *   variables: {
+ *   },
+ * });
+ */
+export function useNotificationMarkAllReadMutation(baseOptions?: Apollo.MutationHookOptions<NotificationMarkAllReadMutation, NotificationMarkAllReadMutationVariables>) {
+        const options = {...defaultOptions, ...baseOptions}
+        return Apollo.useMutation<NotificationMarkAllReadMutation, NotificationMarkAllReadMutationVariables>(NotificationMarkAllReadDocument, options);
+      }
+export type NotificationMarkAllReadMutationHookResult = ReturnType<typeof useNotificationMarkAllReadMutation>;
+export type NotificationMarkAllReadMutationResult = Apollo.MutationResult<NotificationMarkAllReadMutation>;
+export type NotificationMarkAllReadMutationOptions = Apollo.BaseMutationOptions<NotificationMarkAllReadMutation, NotificationMarkAllReadMutationVariables>;
 export const NotificationAddedDocument = gql`
     subscription notificationAdded {
   notificationAdded {

frontend/src/shared/graphql/notifictionMarkAllRead.ts

@@ -0,0 +1,11 @@
+import gql from 'graphql-tag';
+
+const CREATE_TASK_MUTATION = gql`
+  mutation notificationMarkAllRead {
+    notificationMarkAllRead {
+      success
+    }
+  }
+`;
+
+export default CREATE_TASK_MUTATION;

go.mod

@@ -8,6 +8,8 @@ require (
 	github.com/brianvoe/gofakeit/v5 v5.11.2
 	github.com/go-chi/chi v3.3.2+incompatible
 	github.com/go-chi/cors v1.2.0
+	github.com/go-redis/redis v6.15.8+incompatible
+	github.com/go-redis/redis/v8 v8.0.0-beta.6
 	github.com/golang-migrate/migrate/v4 v4.11.0
 	github.com/google/uuid v1.1.1
 	github.com/jinzhu/now v1.1.1

internal/commands/commands.go

@@ -68,6 +68,6 @@ func initConfig() {
 // Execute the root cobra command
 func Execute() {
 	rootCmd.SetVersionTemplate(VersionTemplate())
-	rootCmd.AddCommand(newTokenCmd(), newWebCmd(), newMigrateCmd(), newWorkerCmd(), newResetPasswordCmd(), newSeedCmd())
+	rootCmd.AddCommand(newJobCmd(), newTokenCmd(), newWebCmd(), newMigrateCmd(), newWorkerCmd(), newResetPasswordCmd(), newSeedCmd())
 	rootCmd.Execute()
 }

internal/commands/job.go

@@ -0,0 +1,60 @@
+package commands
+
+import (
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/RichardKnop/machinery/v1"
+	mTasks "github.com/RichardKnop/machinery/v1/tasks"
+
+	queueLog "github.com/RichardKnop/machinery/v1/log"
+	"github.com/jmoiron/sqlx"
+	"github.com/jordanknott/taskcafe/internal/config"
+	"github.com/jordanknott/taskcafe/internal/jobs"
+	log "github.com/sirupsen/logrus"
+)
+
+func newJobCmd() *cobra.Command {
+	cc := &cobra.Command{
+		Use:   "job",
+		Short: "Run a task manually",
+		Long:  "Run a task manually",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			Formatter := new(log.TextFormatter)
+			Formatter.TimestampFormat = "02-01-2006 15:04:05"
+			Formatter.FullTimestamp = true
+			log.SetFormatter(Formatter)
+			log.SetLevel(log.InfoLevel)
+
+			appConfig, err := config.GetAppConfig()
+			if err != nil {
+				log.Panic(err)
+			}
+			db, err := sqlx.Connect("postgres", config.GetDatabaseConfig().GetDatabaseConnectionUri())
+			if err != nil {
+				log.Panic(err)
+			}
+			db.SetMaxOpenConns(25)
+			db.SetMaxIdleConns(25)
+			db.SetConnMaxLifetime(5 * time.Minute)
+			defer db.Close()
+
+			log.Info("starting task queue server instance")
+			jobConfig := appConfig.Job.GetJobConfig()
+			server, err := machinery.NewServer(&jobConfig)
+			if err != nil {
+				// do something with the error
+			}
+			queueLog.Set(&jobs.MachineryLogger{})
+
+			signature := &mTasks.Signature{
+				Name: "scheduleDueDateNotifications",
+			}
+			server.SendTask(signature)
+
+			return nil
+		},
+	}
+	return cc
+}

internal/commands/web.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/RichardKnop/machinery/v1"
+	mTasks "github.com/RichardKnop/machinery/v1/tasks"
 	"github.com/golang-migrate/migrate/v4"
 	"github.com/golang-migrate/migrate/v4/database/postgres"
 	"github.com/golang-migrate/migrate/v4/source/httpfs"
@@ -36,6 +37,12 @@ func newWebCmd() *cobra.Command {
 				return err
 			}
 
+			redisClient, err := appConfig.MessageQueue.GetMessageQueueClient()
+			if err != nil {
+				return err
+			}
+			defer redisClient.Close()
+
 			connection := appConfig.Database.GetDatabaseConnectionUri()
 			var db *sqlx.DB
 			var retryDuration time.Duration
@@ -67,15 +74,17 @@ func newWebCmd() *cobra.Command {
 			}
 
 			var server *machinery.Server
-			if appConfig.Job.Enabled {
+			jobConfig := appConfig.Job.GetJobConfig()
-				jobConfig := appConfig.Job.GetJobConfig()
+			server, err = machinery.NewServer(&jobConfig)
-				server, err = machinery.NewServer(&jobConfig)
+			if err != nil {
-				if err != nil {
+				return err
-					return err
-				}
 			}
+			signature := &mTasks.Signature{
+				Name: "scheduleDueDateNotifications",
+			}
+			server.SendTask(signature)
 
-			r, _ := route.NewRouter(db, server, appConfig)
+			r, _ := route.NewRouter(db, redisClient, server, appConfig)
 			log.WithFields(log.Fields{"url": viper.GetString("server.hostname")}).Info("starting server")
 			return http.ListenAndServe(viper.GetString("server.hostname"), r)
 		},

internal/commands/worker.go

@@ -47,7 +47,11 @@ func newWorkerCmd() *cobra.Command {
 			}
 			queueLog.Set(&jobs.MachineryLogger{})
 			repo := *repo.NewRepository(db)
-			jobs.RegisterTasks(server, repo)
+			redisClient, err := appConfig.MessageQueue.GetMessageQueueClient()
+			if err != nil {
+				return err
+			}
+			jobs.RegisterTasks(server, repo, appConfig, redisClient)
 
 			worker := server.NewWorker("taskcafe_worker", 10)
 			log.Info("starting task queue worker")

internal/config/config.go

@@ -1,10 +1,14 @@
 package config
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
 
+	"github.com/go-redis/redis/v8"
+
 	mConfig "github.com/RichardKnop/machinery/v1/config"
 	"github.com/google/uuid"
 	log "github.com/sirupsen/logrus"
@@ -28,6 +32,8 @@ const (
 	JobStore     = "job.store"
 	JobQueueName = "job.queue_name"
 
+	MessageQueue = "message.queue"
+
 	SmtpFrom       = "smtp.from"
 	SmtpHost       = "smtp.host"
 	SmtpPort       = "smtp.port"
@@ -46,9 +52,10 @@ var defaults = map[string]interface{}{
 	DatabaseSslMode:         "disable",
 	SecurityTokenExpiration: "15m",
 	SecuritySecret:          "",
+	MessageQueue:            "localhost:6379",
 	JobEnabled:              false,
-	JobBroker:               "amqp://guest:guest@localhost:5672/",
+	JobBroker:               "redis://localhost:6379",
-	JobStore:                "memcache://localhost:11211",
+	JobStore:                "redis://localhost:6379",
 	JobQueueName:            "taskcafe_tasks",
 	SmtpFrom:                "no-reply@example.com",
 	SmtpHost:                "localhost",
@@ -65,10 +72,15 @@ func InitDefaults() {
 }
 
 type AppConfig struct {
-	Email    EmailConfig
+	Email        EmailConfig
-	Security SecurityConfig
+	Security     SecurityConfig
-	Database DatabaseConfig
+	Database     DatabaseConfig
-	Job      JobConfig
+	Job          JobConfig
+	MessageQueue MessageQueueConfig
+}
+
+type MessageQueueConfig struct {
+	URI string
 }
 
 type JobConfig struct {
@@ -92,11 +104,12 @@ func (cfg *JobConfig) GetJobConfig() mConfig.Config {
 		Broker:        cfg.Broker,
 		DefaultQueue:  cfg.QueueName,
 		ResultBackend: cfg.Store,
-		AMQP: &mConfig.AMQPConfig{
+		/*
-			Exchange:     "machinery_exchange",
+			AMQP: &mConfig.AMQPConfig{
-			ExchangeType: "direct",
+				Exchange:     "machinery_exchange",
-			BindingKey:   "machinery_task",
+				ExchangeType: "direct",
-		},
+				BindingKey:   "machinery_task",
+			} */
 	}
 }
 
@@ -149,12 +162,14 @@ func GetAppConfig() (AppConfig, error) {
 	jobCfg := GetJobConfig()
 	databaseCfg := GetDatabaseConfig()
 	emailCfg := GetEmailConfig()
+	messageCfg := MessageQueueConfig{URI: viper.GetString("message.queue")}
 	return AppConfig{
-		Email:    emailCfg,
+		Email:        emailCfg,
-		Security: securityCfg,
+		Security:     securityCfg,
-		Database: databaseCfg,
+		Database:     databaseCfg,
-		Job:      jobCfg,
+		Job:          jobCfg,
-	}, nil
+		MessageQueue: messageCfg,
+	}, err
 }
 
 func GetSecurityConfig(accessTokenExp string, secret []byte) (SecurityConfig, error) {
@@ -166,6 +181,19 @@ func GetSecurityConfig(accessTokenExp string, secret []byte) (SecurityConfig, er
 	return SecurityConfig{AccessTokenExpiration: exp, Secret: secret}, nil
 }
 
+func (c MessageQueueConfig) GetMessageQueueClient() (*redis.Client, error) {
+	client := redis.NewClient(&redis.Options{
+		Addr: c.URI,
+	})
+
+	_, err := client.Ping(context.Background()).Result()
+	if !errors.Is(err, nil) {
+		return nil, err
+	}
+
+	return client, nil
+}
+
 func GetEmailConfig() EmailConfig {
 	return EmailConfig{
 		From:               viper.GetString(SmtpFrom),

internal/db/models.go

@@ -192,6 +192,7 @@ type TaskDueDateReminder struct {
 	TaskID            uuid.UUID `json:"task_id"`
 	Period            int32     `json:"period"`
 	Duration          string    `json:"duration"`
+	RemindAt          time.Time `json:"remind_at"`
 }
 
 type TaskDueDateReminderDuration struct {

internal/db/notification.sql.go

@@ -66,9 +66,8 @@ func (q *Queries) CreateNotificationNotifed(ctx context.Context, arg CreateNotif
 }
 
 const getAllNotificationsForUserID = `-- name: GetAllNotificationsForUserID :many
-SELECT notified_id, nn.notification_id, nn.user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on, user_account.user_id, created_at, email, username, password_hash, profile_bg_color, full_name, initials, profile_avatar_url, role_code, bio, active FROM notification_notified AS nn
+SELECT notified_id, nn.notification_id, user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE nn.user_id = $1
 `
 
@@ -83,18 +82,6 @@ type GetAllNotificationsForUserIDRow struct {
 	ActionType       string          `json:"action_type"`
 	Data             json.RawMessage `json:"data"`
 	CreatedOn        time.Time       `json:"created_on"`
-	UserID_2         uuid.UUID       `json:"user_id_2"`
-	CreatedAt        time.Time       `json:"created_at"`
-	Email            string          `json:"email"`
-	Username         string          `json:"username"`
-	PasswordHash     string          `json:"password_hash"`
-	ProfileBgColor   string          `json:"profile_bg_color"`
-	FullName         string          `json:"full_name"`
-	Initials         string          `json:"initials"`
-	ProfileAvatarUrl sql.NullString  `json:"profile_avatar_url"`
-	RoleCode         string          `json:"role_code"`
-	Bio              string          `json:"bio"`
-	Active           bool            `json:"active"`
 }
 
 func (q *Queries) GetAllNotificationsForUserID(ctx context.Context, userID uuid.UUID) ([]GetAllNotificationsForUserIDRow, error) {
@@ -117,18 +104,6 @@ func (q *Queries) GetAllNotificationsForUserID(ctx context.Context, userID uuid.
 			&i.ActionType,
 			&i.Data,
 			&i.CreatedOn,
-			&i.UserID_2,
-			&i.CreatedAt,
-			&i.Email,
-			&i.Username,
-			&i.PasswordHash,
-			&i.ProfileBgColor,
-			&i.FullName,
-			&i.Initials,
-			&i.ProfileAvatarUrl,
-			&i.RoleCode,
-			&i.Bio,
-			&i.Active,
 		); err != nil {
 			return nil, err
 		}
@@ -143,10 +118,26 @@ func (q *Queries) GetAllNotificationsForUserID(ctx context.Context, userID uuid.
 	return items, nil
 }
 
+const getNotificationByID = `-- name: GetNotificationByID :one
+SELECT notification_id, caused_by, action_type, data, created_on FROM notification WHERE notification_id = $1
+`
+
+func (q *Queries) GetNotificationByID(ctx context.Context, notificationID uuid.UUID) (Notification, error) {
+	row := q.db.QueryRowContext(ctx, getNotificationByID, notificationID)
+	var i Notification
+	err := row.Scan(
+		&i.NotificationID,
+		&i.CausedBy,
+		&i.ActionType,
+		&i.Data,
+		&i.CreatedOn,
+	)
+	return i, err
+}
+
 const getNotificationsForUserIDCursor = `-- name: GetNotificationsForUserIDCursor :many
-SELECT notified_id, nn.notification_id, nn.user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on, user_account.user_id, created_at, email, username, password_hash, profile_bg_color, full_name, initials, profile_avatar_url, role_code, bio, active FROM notification_notified AS nn
+SELECT n.notification_id, n.caused_by, n.action_type, n.data, n.created_on, nn.notified_id, nn.notification_id, nn.user_id, nn.read, nn.read_at FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE (n.created_on, n.notification_id) < ($1::timestamptz, $2::uuid)
   AND nn.user_id = $3::uuid
   AND ($4::boolean = false OR nn.read = false)
@@ -166,28 +157,16 @@ type GetNotificationsForUserIDCursorParams struct {
 }
 
 type GetNotificationsForUserIDCursorRow struct {
-	NotifiedID       uuid.UUID       `json:"notified_id"`
 	NotificationID   uuid.UUID       `json:"notification_id"`
-	UserID           uuid.UUID       `json:"user_id"`
-	Read             bool            `json:"read"`
-	ReadAt           sql.NullTime    `json:"read_at"`
-	NotificationID_2 uuid.UUID       `json:"notification_id_2"`
 	CausedBy         uuid.UUID       `json:"caused_by"`
 	ActionType       string          `json:"action_type"`
 	Data             json.RawMessage `json:"data"`
 	CreatedOn        time.Time       `json:"created_on"`
-	UserID_2         uuid.UUID       `json:"user_id_2"`
+	NotifiedID       uuid.UUID       `json:"notified_id"`
-	CreatedAt        time.Time       `json:"created_at"`
+	NotificationID_2 uuid.UUID       `json:"notification_id_2"`
-	Email            string          `json:"email"`
+	UserID           uuid.UUID       `json:"user_id"`
-	Username         string          `json:"username"`
+	Read             bool            `json:"read"`
-	PasswordHash     string          `json:"password_hash"`
+	ReadAt           sql.NullTime    `json:"read_at"`
-	ProfileBgColor   string          `json:"profile_bg_color"`
-	FullName         string          `json:"full_name"`
-	Initials         string          `json:"initials"`
-	ProfileAvatarUrl sql.NullString  `json:"profile_avatar_url"`
-	RoleCode         string          `json:"role_code"`
-	Bio              string          `json:"bio"`
-	Active           bool            `json:"active"`
 }
 
 func (q *Queries) GetNotificationsForUserIDCursor(ctx context.Context, arg GetNotificationsForUserIDCursorParams) ([]GetNotificationsForUserIDCursorRow, error) {
@@ -208,28 +187,16 @@ func (q *Queries) GetNotificationsForUserIDCursor(ctx context.Context, arg GetNo
 	for rows.Next() {
 		var i GetNotificationsForUserIDCursorRow
 		if err := rows.Scan(
-			&i.NotifiedID,
 			&i.NotificationID,
-			&i.UserID,
-			&i.Read,
-			&i.ReadAt,
-			&i.NotificationID_2,
 			&i.CausedBy,
 			&i.ActionType,
 			&i.Data,
 			&i.CreatedOn,
-			&i.UserID_2,
+			&i.NotifiedID,
-			&i.CreatedAt,
+			&i.NotificationID_2,
-			&i.Email,
+			&i.UserID,
-			&i.Username,
+			&i.Read,
-			&i.PasswordHash,
+			&i.ReadAt,
-			&i.ProfileBgColor,
-			&i.FullName,
-			&i.Initials,
-			&i.ProfileAvatarUrl,
-			&i.RoleCode,
-			&i.Bio,
-			&i.Active,
 		); err != nil {
 			return nil, err
 		}
@@ -245,9 +212,8 @@ func (q *Queries) GetNotificationsForUserIDCursor(ctx context.Context, arg GetNo
 }
 
 const getNotificationsForUserIDPaged = `-- name: GetNotificationsForUserIDPaged :many
-SELECT notified_id, nn.notification_id, nn.user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on, user_account.user_id, created_at, email, username, password_hash, profile_bg_color, full_name, initials, profile_avatar_url, role_code, bio, active FROM notification_notified AS nn
+SELECT n.notification_id, n.caused_by, n.action_type, n.data, n.created_on, nn.notified_id, nn.notification_id, nn.user_id, nn.read, nn.read_at FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE nn.user_id = $1::uuid
   AND ($2::boolean = false OR nn.read = false)
   AND ($3::boolean = false OR n.action_type = ANY($4::text[]))
@@ -264,28 +230,16 @@ type GetNotificationsForUserIDPagedParams struct {
 }
 
 type GetNotificationsForUserIDPagedRow struct {
-	NotifiedID       uuid.UUID       `json:"notified_id"`
 	NotificationID   uuid.UUID       `json:"notification_id"`
-	UserID           uuid.UUID       `json:"user_id"`
-	Read             bool            `json:"read"`
-	ReadAt           sql.NullTime    `json:"read_at"`
-	NotificationID_2 uuid.UUID       `json:"notification_id_2"`
 	CausedBy         uuid.UUID       `json:"caused_by"`
 	ActionType       string          `json:"action_type"`
 	Data             json.RawMessage `json:"data"`
 	CreatedOn        time.Time       `json:"created_on"`
-	UserID_2         uuid.UUID       `json:"user_id_2"`
+	NotifiedID       uuid.UUID       `json:"notified_id"`
-	CreatedAt        time.Time       `json:"created_at"`
+	NotificationID_2 uuid.UUID       `json:"notification_id_2"`
-	Email            string          `json:"email"`
+	UserID           uuid.UUID       `json:"user_id"`
-	Username         string          `json:"username"`
+	Read             bool            `json:"read"`
-	PasswordHash     string          `json:"password_hash"`
+	ReadAt           sql.NullTime    `json:"read_at"`
-	ProfileBgColor   string          `json:"profile_bg_color"`
-	FullName         string          `json:"full_name"`
-	Initials         string          `json:"initials"`
-	ProfileAvatarUrl sql.NullString  `json:"profile_avatar_url"`
-	RoleCode         string          `json:"role_code"`
-	Bio              string          `json:"bio"`
-	Active           bool            `json:"active"`
 }
 
 func (q *Queries) GetNotificationsForUserIDPaged(ctx context.Context, arg GetNotificationsForUserIDPagedParams) ([]GetNotificationsForUserIDPagedRow, error) {
@@ -304,28 +258,16 @@ func (q *Queries) GetNotificationsForUserIDPaged(ctx context.Context, arg GetNot
 	for rows.Next() {
 		var i GetNotificationsForUserIDPagedRow
 		if err := rows.Scan(
-			&i.NotifiedID,
 			&i.NotificationID,
-			&i.UserID,
-			&i.Read,
-			&i.ReadAt,
-			&i.NotificationID_2,
 			&i.CausedBy,
 			&i.ActionType,
 			&i.Data,
 			&i.CreatedOn,
-			&i.UserID_2,
+			&i.NotifiedID,
-			&i.CreatedAt,
+			&i.NotificationID_2,
-			&i.Email,
+			&i.UserID,
-			&i.Username,
+			&i.Read,
-			&i.PasswordHash,
+			&i.ReadAt,
-			&i.ProfileBgColor,
-			&i.FullName,
-			&i.Initials,
-			&i.ProfileAvatarUrl,
-			&i.RoleCode,
-			&i.Bio,
-			&i.Active,
 		); err != nil {
 			return nil, err
 		}
@@ -341,9 +283,8 @@ func (q *Queries) GetNotificationsForUserIDPaged(ctx context.Context, arg GetNot
 }
 
 const getNotifiedByID = `-- name: GetNotifiedByID :one
-SELECT notified_id, nn.notification_id, nn.user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on, user_account.user_id, created_at, email, username, password_hash, profile_bg_color, full_name, initials, profile_avatar_url, role_code, bio, active FROM notification_notified as nn
+SELECT notified_id, nn.notification_id, user_id, read, read_at, n.notification_id, caused_by, action_type, data, created_on FROM notification_notified as nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE notified_id = $1
 `
 
@@ -358,18 +299,6 @@ type GetNotifiedByIDRow struct {
 	ActionType       string          `json:"action_type"`
 	Data             json.RawMessage `json:"data"`
 	CreatedOn        time.Time       `json:"created_on"`
-	UserID_2         uuid.UUID       `json:"user_id_2"`
-	CreatedAt        time.Time       `json:"created_at"`
-	Email            string          `json:"email"`
-	Username         string          `json:"username"`
-	PasswordHash     string          `json:"password_hash"`
-	ProfileBgColor   string          `json:"profile_bg_color"`
-	FullName         string          `json:"full_name"`
-	Initials         string          `json:"initials"`
-	ProfileAvatarUrl sql.NullString  `json:"profile_avatar_url"`
-	RoleCode         string          `json:"role_code"`
-	Bio              string          `json:"bio"`
-	Active           bool            `json:"active"`
 }
 
 func (q *Queries) GetNotifiedByID(ctx context.Context, notifiedID uuid.UUID) (GetNotifiedByIDRow, error) {
@@ -386,18 +315,23 @@ func (q *Queries) GetNotifiedByID(ctx context.Context, notifiedID uuid.UUID) (Ge
 		&i.ActionType,
 		&i.Data,
 		&i.CreatedOn,
-		&i.UserID_2,
+	)
-		&i.CreatedAt,
+	return i, err
-		&i.Email,
+}
-		&i.Username,
+
-		&i.PasswordHash,
+const getNotifiedByIDNoExtra = `-- name: GetNotifiedByIDNoExtra :one
-		&i.ProfileBgColor,
+SELECT notified_id, notification_id, user_id, read, read_at FROM notification_notified as nn WHERE nn.notified_id = $1
-		&i.FullName,
+`
-		&i.Initials,
+
-		&i.ProfileAvatarUrl,
+func (q *Queries) GetNotifiedByIDNoExtra(ctx context.Context, notifiedID uuid.UUID) (NotificationNotified, error) {
-		&i.RoleCode,
+	row := q.db.QueryRowContext(ctx, getNotifiedByIDNoExtra, notifiedID)
-		&i.Bio,
+	var i NotificationNotified
-		&i.Active,
+	err := row.Scan(
+		&i.NotifiedID,
+		&i.NotificationID,
+		&i.UserID,
+		&i.Read,
+		&i.ReadAt,
 	)
 	return i, err
 }
@@ -413,6 +347,20 @@ func (q *Queries) HasUnreadNotification(ctx context.Context, userID uuid.UUID) (
 	return exists, err
 }
 
+const markAllNotificationsRead = `-- name: MarkAllNotificationsRead :exec
+UPDATE notification_notified SET read = true, read_at = $2 WHERE user_id = $1
+`
+
+type MarkAllNotificationsReadParams struct {
+	UserID uuid.UUID    `json:"user_id"`
+	ReadAt sql.NullTime `json:"read_at"`
+}
+
+func (q *Queries) MarkAllNotificationsRead(ctx context.Context, arg MarkAllNotificationsReadParams) error {
+	_, err := q.db.ExecContext(ctx, markAllNotificationsRead, arg.UserID, arg.ReadAt)
+	return err
+}
+
 const markNotificationAsRead = `-- name: MarkNotificationAsRead :exec
 UPDATE notification_notified SET read = $3, read_at = $2 WHERE user_id = $1 AND notified_id = $4
 `

internal/db/querier.go

@@ -5,6 +5,7 @@ package db
 import (
 	"context"
 	"database/sql"
+	"time"
 
 	"github.com/google/uuid"
 )
@@ -82,6 +83,8 @@ type Querier interface {
 	GetCommentsForTaskID(ctx context.Context, taskID uuid.UUID) ([]TaskComment, error)
 	GetConfirmTokenByEmail(ctx context.Context, email string) (UserAccountConfirmToken, error)
 	GetConfirmTokenByID(ctx context.Context, confirmTokenID uuid.UUID) (UserAccountConfirmToken, error)
+	GetDueDateReminderByID(ctx context.Context, dueDateReminderID uuid.UUID) (TaskDueDateReminder, error)
+	GetDueDateRemindersForDuration(ctx context.Context, startAt time.Time) ([]TaskDueDateReminder, error)
 	GetDueDateRemindersForTaskID(ctx context.Context, taskID uuid.UUID) ([]TaskDueDateReminder, error)
 	GetInvitedMembersForProjectID(ctx context.Context, projectID uuid.UUID) ([]GetInvitedMembersForProjectIDRow, error)
 	GetInvitedUserAccounts(ctx context.Context) ([]UserAccountInvited, error)
@@ -92,9 +95,11 @@ type Querier interface {
 	GetMemberData(ctx context.Context, projectID uuid.UUID) ([]UserAccount, error)
 	GetMemberProjectIDsForUserID(ctx context.Context, userID uuid.UUID) ([]uuid.UUID, error)
 	GetMemberTeamIDsForUserID(ctx context.Context, userID uuid.UUID) ([]uuid.UUID, error)
+	GetNotificationByID(ctx context.Context, notificationID uuid.UUID) (Notification, error)
 	GetNotificationsForUserIDCursor(ctx context.Context, arg GetNotificationsForUserIDCursorParams) ([]GetNotificationsForUserIDCursorRow, error)
 	GetNotificationsForUserIDPaged(ctx context.Context, arg GetNotificationsForUserIDPagedParams) ([]GetNotificationsForUserIDPagedRow, error)
 	GetNotifiedByID(ctx context.Context, notifiedID uuid.UUID) (GetNotifiedByIDRow, error)
+	GetNotifiedByIDNoExtra(ctx context.Context, notifiedID uuid.UUID) (NotificationNotified, error)
 	GetPersonalProjectsForUserID(ctx context.Context, userID uuid.UUID) ([]Project, error)
 	GetProjectByID(ctx context.Context, projectID uuid.UUID) (Project, error)
 	GetProjectIDByShortID(ctx context.Context, shortID string) (uuid.UUID, error)
@@ -121,6 +126,7 @@ type Querier interface {
 	GetTaskChecklistItemByID(ctx context.Context, taskChecklistItemID uuid.UUID) (TaskChecklistItem, error)
 	GetTaskChecklistItemsForTaskChecklist(ctx context.Context, taskChecklistID uuid.UUID) ([]TaskChecklistItem, error)
 	GetTaskChecklistsForTask(ctx context.Context, taskID uuid.UUID) ([]TaskChecklist, error)
+	GetTaskForDueDateReminder(ctx context.Context, dueDateReminderID uuid.UUID) (Task, error)
 	GetTaskGroupByID(ctx context.Context, taskGroupID uuid.UUID) (TaskGroup, error)
 	GetTaskGroupsForProject(ctx context.Context, projectID uuid.UUID) ([]TaskGroup, error)
 	GetTaskIDByShortID(ctx context.Context, shortID string) (uuid.UUID, error)
@@ -128,6 +134,7 @@ type Querier interface {
 	GetTaskLabelForTaskByProjectLabelID(ctx context.Context, arg GetTaskLabelForTaskByProjectLabelIDParams) (TaskLabel, error)
 	GetTaskLabelsForTaskID(ctx context.Context, taskID uuid.UUID) ([]TaskLabel, error)
 	GetTaskWatcher(ctx context.Context, arg GetTaskWatcherParams) (TaskWatcher, error)
+	GetTaskWatchersForTask(ctx context.Context, taskID uuid.UUID) ([]TaskWatcher, error)
 	GetTasksForTaskGroupID(ctx context.Context, taskGroupID uuid.UUID) ([]Task, error)
 	GetTeamByID(ctx context.Context, teamID uuid.UUID) (Team, error)
 	GetTeamMemberByID(ctx context.Context, arg GetTeamMemberByIDParams) (TeamMember, error)
@@ -144,6 +151,7 @@ type Querier interface {
 	HasActiveUser(ctx context.Context) (bool, error)
 	HasAnyUser(ctx context.Context) (bool, error)
 	HasUnreadNotification(ctx context.Context, userID uuid.UUID) (bool, error)
+	MarkAllNotificationsRead(ctx context.Context, arg MarkAllNotificationsReadParams) error
 	MarkNotificationAsRead(ctx context.Context, arg MarkNotificationAsReadParams) error
 	SetFirstUserActive(ctx context.Context) (UserAccount, error)
 	SetInactiveLastMoveForTaskID(ctx context.Context, taskID uuid.UUID) error
@@ -154,6 +162,7 @@ type Querier interface {
 	SetUserActiveByEmail(ctx context.Context, email string) (UserAccount, error)
 	SetUserPassword(ctx context.Context, arg SetUserPasswordParams) (UserAccount, error)
 	UpdateDueDateReminder(ctx context.Context, arg UpdateDueDateReminderParams) (TaskDueDateReminder, error)
+	UpdateDueDateReminderRemindAt(ctx context.Context, arg UpdateDueDateReminderRemindAtParams) (TaskDueDateReminder, error)
 	UpdateProjectLabel(ctx context.Context, arg UpdateProjectLabelParams) (ProjectLabel, error)
 	UpdateProjectLabelColor(ctx context.Context, arg UpdateProjectLabelColorParams) (ProjectLabel, error)
 	UpdateProjectLabelName(ctx context.Context, arg UpdateProjectLabelNameParams) (ProjectLabel, error)

internal/db/query/notification.sql

@@ -1,21 +1,25 @@
 -- name: GetAllNotificationsForUserID :many
 SELECT * FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE nn.user_id = $1;
 
 -- name: GetNotifiedByID :one
 SELECT * FROM notification_notified as nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE notified_id = $1;
 
+-- name: GetNotifiedByIDNoExtra :one
+SELECT * FROM notification_notified as nn WHERE nn.notified_id = $1;
+
 -- name: HasUnreadNotification :one
 SELECT EXISTS (SELECT 1 FROM notification_notified WHERE read = false AND user_id = $1);
 
 -- name: MarkNotificationAsRead :exec
 UPDATE notification_notified SET read = $3, read_at = $2 WHERE user_id = $1 AND notified_id = $4;
 
+-- name: MarkAllNotificationsRead :exec
+UPDATE notification_notified SET read = true, read_at = $2 WHERE user_id = $1;
+
 -- name: CreateNotification :one
 INSERT INTO notification (caused_by, data, action_type, created_on)
   VALUES ($1, $2, $3, $4) RETURNING *;
@@ -23,10 +27,12 @@ INSERT INTO notification (caused_by, data, action_type, created_on)
 -- name: CreateNotificationNotifed :one
 INSERT INTO notification_notified (notification_id, user_id) VALUES ($1, $2) RETURNING *;
 
+-- name: GetNotificationByID :one
+SELECT * FROM notification WHERE notification_id = $1;
+
 -- name: GetNotificationsForUserIDPaged :many
-SELECT * FROM notification_notified AS nn
+SELECT n.*, nn.* FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE nn.user_id = @user_id::uuid
   AND (@enable_unread::boolean = false OR nn.read = false)
   AND (@enable_action_type::boolean = false OR n.action_type = ANY(@action_type::text[]))
@@ -34,9 +40,8 @@ SELECT * FROM notification_notified AS nn
   LIMIT @limit_rows::int;
 
 -- name: GetNotificationsForUserIDCursor :many
-SELECT * FROM notification_notified AS nn
+SELECT n.*, nn.* FROM notification_notified AS nn
   INNER JOIN notification AS n ON n.notification_id = nn.notification_id
-  LEFT JOIN user_account ON user_account.user_id = n.caused_by
   WHERE (n.created_on, n.notification_id) < (@created_on::timestamptz, @notification_id::uuid)
   AND nn.user_id = @user_id::uuid
   AND (@enable_unread::boolean = false OR nn.read = false)

internal/db/query/task.sql

@@ -1,6 +1,9 @@
 -- name: GetTaskWatcher :one
 SELECT * FROM task_watcher WHERE user_id = $1 AND task_id = $2;
 
+-- name: GetTaskWatchersForTask :many
+SELECT * FROM task_watcher WHERE task_id = $1;
+
 -- name: CreateTaskWatcher :one
 INSERT INTO task_watcher (user_id, task_id, watched_at) VALUES ($1, $2, $3) RETURNING *;
 
@@ -119,13 +122,28 @@ SELECT COUNT(*) FROM task_comment WHERE task_id = $1;
 
 
 -- name: CreateDueDateReminder :one
-INSERT INTO task_due_date_reminder (task_id, period, duration) VALUES ($1, $2, $3) RETURNING *;
+INSERT INTO task_due_date_reminder (task_id, period, duration, remind_at) VALUES ($1, $2, $3, $4) RETURNING *;
 
 -- name: UpdateDueDateReminder :one
-UPDATE task_due_date_reminder SET period = $2, duration = $3 WHERE due_date_reminder_id = $1 RETURNING *;
+UPDATE task_due_date_reminder SET remind_at = $4, period = $2, duration = $3 WHERE due_date_reminder_id = $1 RETURNING *;
+
+-- name: GetTaskForDueDateReminder :one
+SELECT task.* FROM task_due_date_reminder 
+  INNER JOIN task ON task.task_id = task_due_date_reminder.task_id
+  WHERE task_due_date_reminder.due_date_reminder_id = $1;
+
+-- name: UpdateDueDateReminderRemindAt :one
+UPDATE task_due_date_reminder SET remind_at = $2 WHERE due_date_reminder_id = $1 RETURNING *;
 
 -- name: GetDueDateRemindersForTaskID :many
 SELECT * FROM task_due_date_reminder WHERE task_id = $1;
 
+-- name: GetDueDateReminderByID :one
+SELECT * FROM task_due_date_reminder WHERE due_date_reminder_id = $1;
+
 -- name: DeleteDueDateReminder :exec
 DELETE FROM task_due_date_reminder WHERE due_date_reminder_id = $1;
+
+-- name: GetDueDateRemindersForDuration :many
+SELECT * FROM task_due_date_reminder WHERE remind_at >= @start_at::timestamptz;
+

internal/db/task.sql.go

@@ -13,23 +13,30 @@ import (
 )
 
 const createDueDateReminder = `-- name: CreateDueDateReminder :one
-INSERT INTO task_due_date_reminder (task_id, period, duration) VALUES ($1, $2, $3) RETURNING due_date_reminder_id, task_id, period, duration
+INSERT INTO task_due_date_reminder (task_id, period, duration, remind_at) VALUES ($1, $2, $3, $4) RETURNING due_date_reminder_id, task_id, period, duration, remind_at
 `
 
 type CreateDueDateReminderParams struct {
 	TaskID   uuid.UUID `json:"task_id"`
 	Period   int32     `json:"period"`
 	Duration string    `json:"duration"`
+	RemindAt time.Time `json:"remind_at"`
 }
 
 func (q *Queries) CreateDueDateReminder(ctx context.Context, arg CreateDueDateReminderParams) (TaskDueDateReminder, error) {
-	row := q.db.QueryRowContext(ctx, createDueDateReminder, arg.TaskID, arg.Period, arg.Duration)
+	row := q.db.QueryRowContext(ctx, createDueDateReminder,
+		arg.TaskID,
+		arg.Period,
+		arg.Duration,
+		arg.RemindAt,
+	)
 	var i TaskDueDateReminder
 	err := row.Scan(
 		&i.DueDateReminderID,
 		&i.TaskID,
 		&i.Period,
 		&i.Duration,
+		&i.RemindAt,
 	)
 	return i, err
 }
@@ -434,8 +441,58 @@ func (q *Queries) GetCommentsForTaskID(ctx context.Context, taskID uuid.UUID) ([
 	return items, nil
 }
 
+const getDueDateReminderByID = `-- name: GetDueDateReminderByID :one
+SELECT due_date_reminder_id, task_id, period, duration, remind_at FROM task_due_date_reminder WHERE due_date_reminder_id = $1
+`
+
+func (q *Queries) GetDueDateReminderByID(ctx context.Context, dueDateReminderID uuid.UUID) (TaskDueDateReminder, error) {
+	row := q.db.QueryRowContext(ctx, getDueDateReminderByID, dueDateReminderID)
+	var i TaskDueDateReminder
+	err := row.Scan(
+		&i.DueDateReminderID,
+		&i.TaskID,
+		&i.Period,
+		&i.Duration,
+		&i.RemindAt,
+	)
+	return i, err
+}
+
+const getDueDateRemindersForDuration = `-- name: GetDueDateRemindersForDuration :many
+SELECT due_date_reminder_id, task_id, period, duration, remind_at FROM task_due_date_reminder WHERE remind_at >= $1::timestamptz
+`
+
+func (q *Queries) GetDueDateRemindersForDuration(ctx context.Context, startAt time.Time) ([]TaskDueDateReminder, error) {
+	rows, err := q.db.QueryContext(ctx, getDueDateRemindersForDuration, startAt)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []TaskDueDateReminder
+	for rows.Next() {
+		var i TaskDueDateReminder
+		if err := rows.Scan(
+			&i.DueDateReminderID,
+			&i.TaskID,
+			&i.Period,
+			&i.Duration,
+			&i.RemindAt,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getDueDateRemindersForTaskID = `-- name: GetDueDateRemindersForTaskID :many
-SELECT due_date_reminder_id, task_id, period, duration FROM task_due_date_reminder WHERE task_id = $1
+SELECT due_date_reminder_id, task_id, period, duration, remind_at FROM task_due_date_reminder WHERE task_id = $1
 `
 
 func (q *Queries) GetDueDateRemindersForTaskID(ctx context.Context, taskID uuid.UUID) ([]TaskDueDateReminder, error) {
@@ -452,6 +509,7 @@ func (q *Queries) GetDueDateRemindersForTaskID(ctx context.Context, taskID uuid.
 			&i.TaskID,
 			&i.Period,
 			&i.Duration,
+			&i.RemindAt,
 		); err != nil {
 			return nil, err
 		}
@@ -614,6 +672,31 @@ func (q *Queries) GetTaskByID(ctx context.Context, taskID uuid.UUID) (Task, erro
 	return i, err
 }
 
+const getTaskForDueDateReminder = `-- name: GetTaskForDueDateReminder :one
+SELECT task.task_id, task.task_group_id, task.created_at, task.name, task.position, task.description, task.due_date, task.complete, task.completed_at, task.has_time, task.short_id FROM task_due_date_reminder 
+  INNER JOIN task ON task.task_id = task_due_date_reminder.task_id
+  WHERE task_due_date_reminder.due_date_reminder_id = $1
+`
+
+func (q *Queries) GetTaskForDueDateReminder(ctx context.Context, dueDateReminderID uuid.UUID) (Task, error) {
+	row := q.db.QueryRowContext(ctx, getTaskForDueDateReminder, dueDateReminderID)
+	var i Task
+	err := row.Scan(
+		&i.TaskID,
+		&i.TaskGroupID,
+		&i.CreatedAt,
+		&i.Name,
+		&i.Position,
+		&i.Description,
+		&i.DueDate,
+		&i.Complete,
+		&i.CompletedAt,
+		&i.HasTime,
+		&i.ShortID,
+	)
+	return i, err
+}
+
 const getTaskIDByShortID = `-- name: GetTaskIDByShortID :one
 SELECT task_id FROM task WHERE short_id = $1
 `
@@ -646,6 +729,38 @@ func (q *Queries) GetTaskWatcher(ctx context.Context, arg GetTaskWatcherParams)
 	return i, err
 }
 
+const getTaskWatchersForTask = `-- name: GetTaskWatchersForTask :many
+SELECT task_watcher_id, task_id, user_id, watched_at FROM task_watcher WHERE task_id = $1
+`
+
+func (q *Queries) GetTaskWatchersForTask(ctx context.Context, taskID uuid.UUID) ([]TaskWatcher, error) {
+	rows, err := q.db.QueryContext(ctx, getTaskWatchersForTask, taskID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var items []TaskWatcher
+	for rows.Next() {
+		var i TaskWatcher
+		if err := rows.Scan(
+			&i.TaskWatcherID,
+			&i.TaskID,
+			&i.UserID,
+			&i.WatchedAt,
+		); err != nil {
+			return nil, err
+		}
+		items = append(items, i)
+	}
+	if err := rows.Close(); err != nil {
+		return nil, err
+	}
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	return items, nil
+}
+
 const getTasksForTaskGroupID = `-- name: GetTasksForTaskGroupID :many
 SELECT task_id, task_group_id, created_at, name, position, description, due_date, complete, completed_at, has_time, short_id FROM task WHERE task_group_id = $1
 `
@@ -715,23 +830,52 @@ func (q *Queries) SetTaskComplete(ctx context.Context, arg SetTaskCompleteParams
 }
 
 const updateDueDateReminder = `-- name: UpdateDueDateReminder :one
-UPDATE task_due_date_reminder SET period = $2, duration = $3 WHERE due_date_reminder_id = $1 RETURNING due_date_reminder_id, task_id, period, duration
+UPDATE task_due_date_reminder SET remind_at = $4, period = $2, duration = $3 WHERE due_date_reminder_id = $1 RETURNING due_date_reminder_id, task_id, period, duration, remind_at
 `
 
 type UpdateDueDateReminderParams struct {
 	DueDateReminderID uuid.UUID `json:"due_date_reminder_id"`
 	Period            int32     `json:"period"`
 	Duration          string    `json:"duration"`
+	RemindAt          time.Time `json:"remind_at"`
 }
 
 func (q *Queries) UpdateDueDateReminder(ctx context.Context, arg UpdateDueDateReminderParams) (TaskDueDateReminder, error) {
-	row := q.db.QueryRowContext(ctx, updateDueDateReminder, arg.DueDateReminderID, arg.Period, arg.Duration)
+	row := q.db.QueryRowContext(ctx, updateDueDateReminder,
+		arg.DueDateReminderID,
+		arg.Period,
+		arg.Duration,
+		arg.RemindAt,
+	)
 	var i TaskDueDateReminder
 	err := row.Scan(
 		&i.DueDateReminderID,
 		&i.TaskID,
 		&i.Period,
 		&i.Duration,
+		&i.RemindAt,
+	)
+	return i, err
+}
+
+const updateDueDateReminderRemindAt = `-- name: UpdateDueDateReminderRemindAt :one
+UPDATE task_due_date_reminder SET remind_at = $2 WHERE due_date_reminder_id = $1 RETURNING due_date_reminder_id, task_id, period, duration, remind_at
+`
+
+type UpdateDueDateReminderRemindAtParams struct {
+	DueDateReminderID uuid.UUID `json:"due_date_reminder_id"`
+	RemindAt          time.Time `json:"remind_at"`
+}
+
+func (q *Queries) UpdateDueDateReminderRemindAt(ctx context.Context, arg UpdateDueDateReminderRemindAtParams) (TaskDueDateReminder, error) {
+	row := q.db.QueryRowContext(ctx, updateDueDateReminderRemindAt, arg.DueDateReminderID, arg.RemindAt)
+	var i TaskDueDateReminder
+	err := row.Scan(
+		&i.DueDateReminderID,
+		&i.TaskID,
+		&i.Period,
+		&i.Duration,
+		&i.RemindAt,
 	)
 	return i, err
 }

internal/graph/generated.go

@@ -277,6 +277,7 @@ type ComplexityRoot struct {
 		DuplicateTaskGroup              func(childComplexity int, input DuplicateTaskGroup) int
 		InviteProjectMembers            func(childComplexity int, input InviteProjectMembers) int
 		LogoutUser                      func(childComplexity int, input LogoutUser) int
+		NotificationMarkAllRead         func(childComplexity int) int
 		NotificationToggleRead          func(childComplexity int, input NotificationToggleReadInput) int
 		RemoveTaskLabel                 func(childComplexity int, input *RemoveTaskLabelInput) int
 		SetTaskChecklistItemComplete    func(childComplexity int, input SetTaskChecklistItemComplete) int
@@ -333,6 +334,10 @@ type ComplexityRoot struct {
 		Value func(childComplexity int) int
 	}
 
+	NotificationMarkAllAsReadResult struct {
+		Success func(childComplexity int) int
+	}
+
 	Notified struct {
 		ID           func(childComplexity int) int
 		Notification func(childComplexity int) int
@@ -617,6 +622,7 @@ type LabelColorResolver interface {
 }
 type MutationResolver interface {
 	NotificationToggleRead(ctx context.Context, input NotificationToggleReadInput) (*Notified, error)
+	NotificationMarkAllRead(ctx context.Context) (*NotificationMarkAllAsReadResult, error)
 	CreateProjectLabel(ctx context.Context, input NewProjectLabel) (*db.ProjectLabel, error)
 	DeleteProjectLabel(ctx context.Context, input DeleteProjectLabel) (*db.ProjectLabel, error)
 	UpdateProjectLabel(ctx context.Context, input UpdateProjectLabel) (*db.ProjectLabel, error)
@@ -1755,6 +1761,13 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in
 
 		return e.complexity.Mutation.LogoutUser(childComplexity, args["input"].(LogoutUser)), true
 
+	case "Mutation.notificationMarkAllRead":
+		if e.complexity.Mutation.NotificationMarkAllRead == nil {
+			break
+		}
+
+		return e.complexity.Mutation.NotificationMarkAllRead(childComplexity), true
+
 	case "Mutation.notificationToggleRead":
 		if e.complexity.Mutation.NotificationToggleRead == nil {
 			break
@@ -2199,6 +2212,13 @@ func (e *executableSchema) Complexity(typeName, field string, childComplexity in
 
 		return e.complexity.NotificationData.Value(childComplexity), true
 
+	case "NotificationMarkAllAsReadResult.success":
+		if e.complexity.NotificationMarkAllAsReadResult.Success == nil {
+			break
+		}
+
+		return e.complexity.NotificationMarkAllAsReadResult.Success(childComplexity), true
+
 	case "Notified.id":
 		if e.complexity.Notified.ID == nil {
 			break
@@ -3417,6 +3437,10 @@ extend type Query {
 
 extend type Mutation {
   notificationToggleRead(input: NotificationToggleReadInput!): Notified!
+  notificationMarkAllRead: NotificationMarkAllAsReadResult!
+}
+type NotificationMarkAllAsReadResult {
+  success: Boolean!
 }
 
 type HasUnreadNotificationsResult {
@@ -3452,6 +3476,7 @@ enum ActionType {
   DUE_DATE_ADDED
   DUE_DATE_REMOVED
   DUE_DATE_CHANGED
+  DUE_DATE_REMINDER
   TASK_ASSIGNED
   TASK_MOVED
   TASK_ARCHIVED
@@ -8535,6 +8560,41 @@ func (ec *executionContext) _Mutation_notificationToggleRead(ctx context.Context
 	return ec.marshalNNotified2ᚖgithubᚗcomᚋjordanknottᚋtaskcafeᚋinternalᚋgraphᚐNotified(ctx, field.Selections, res)
 }
 
+func (ec *executionContext) _Mutation_notificationMarkAllRead(ctx context.Context, field graphql.CollectedField) (ret graphql.Marshaler) {
+	defer func() {
+		if r := recover(); r != nil {
+			ec.Error(ctx, ec.Recover(ctx, r))
+			ret = graphql.Null
+		}
+	}()
+	fc := &graphql.FieldContext{
+		Object:     "Mutation",
+		Field:      field,
+		Args:       nil,
+		IsMethod:   true,
+		IsResolver: true,
+	}
+
+	ctx = graphql.WithFieldContext(ctx, fc)
+	resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) {
+		ctx = rctx // use context from middleware stack in children
+		return ec.resolvers.Mutation().NotificationMarkAllRead(rctx)
+	})
+	if err != nil {
+		ec.Error(ctx, err)
+		return graphql.Null
+	}
+	if resTmp == nil {
+		if !graphql.HasFieldError(ctx, fc) {
+			ec.Errorf(ctx, "must not be null")
+		}
+		return graphql.Null
+	}
+	res := resTmp.(*NotificationMarkAllAsReadResult)
+	fc.Result = res
+	return ec.marshalNNotificationMarkAllAsReadResult2ᚖgithubᚗcomᚋjordanknottᚋtaskcafeᚋinternalᚋgraphᚐNotificationMarkAllAsReadResult(ctx, field.Selections, res)
+}
+
 func (ec *executionContext) _Mutation_createProjectLabel(ctx context.Context, field graphql.CollectedField) (ret graphql.Marshaler) {
 	defer func() {
 		if r := recover(); r != nil {
@@ -13267,6 +13327,41 @@ func (ec *executionContext) _NotificationData_value(ctx context.Context, field g
 	return ec.marshalNString2string(ctx, field.Selections, res)
 }
 
+func (ec *executionContext) _NotificationMarkAllAsReadResult_success(ctx context.Context, field graphql.CollectedField, obj *NotificationMarkAllAsReadResult) (ret graphql.Marshaler) {
+	defer func() {
+		if r := recover(); r != nil {
+			ec.Error(ctx, ec.Recover(ctx, r))
+			ret = graphql.Null
+		}
+	}()
+	fc := &graphql.FieldContext{
+		Object:     "NotificationMarkAllAsReadResult",
+		Field:      field,
+		Args:       nil,
+		IsMethod:   false,
+		IsResolver: false,
+	}
+
+	ctx = graphql.WithFieldContext(ctx, fc)
+	resTmp, err := ec.ResolverMiddleware(ctx, func(rctx context.Context) (interface{}, error) {
+		ctx = rctx // use context from middleware stack in children
+		return obj.Success, nil
+	})
+	if err != nil {
+		ec.Error(ctx, err)
+		return graphql.Null
+	}
+	if resTmp == nil {
+		if !graphql.HasFieldError(ctx, fc) {
+			ec.Errorf(ctx, "must not be null")
+		}
+		return graphql.Null
+	}
+	res := resTmp.(bool)
+	fc.Result = res
+	return ec.marshalNBoolean2bool(ctx, field.Selections, res)
+}
+
 func (ec *executionContext) _Notified_id(ctx context.Context, field graphql.CollectedField, obj *Notified) (ret graphql.Marshaler) {
 	defer func() {
 		if r := recover(); r != nil {
@@ -23074,6 +23169,11 @@ func (ec *executionContext) _Mutation(ctx context.Context, sel ast.SelectionSet)
 			if out.Values[i] == graphql.Null {
 				invalids++
 			}
+		case "notificationMarkAllRead":
+			out.Values[i] = ec._Mutation_notificationMarkAllRead(ctx, field)
+			if out.Values[i] == graphql.Null {
+				invalids++
+			}
 		case "createProjectLabel":
 			out.Values[i] = ec._Mutation_createProjectLabel(ctx, field)
 			if out.Values[i] == graphql.Null {
@@ -23580,6 +23680,33 @@ func (ec *executionContext) _NotificationData(ctx context.Context, sel ast.Selec
 	return out
 }
 
+var notificationMarkAllAsReadResultImplementors = []string{"NotificationMarkAllAsReadResult"}
+
+func (ec *executionContext) _NotificationMarkAllAsReadResult(ctx context.Context, sel ast.SelectionSet, obj *NotificationMarkAllAsReadResult) graphql.Marshaler {
+	fields := graphql.CollectFields(ec.OperationContext, sel, notificationMarkAllAsReadResultImplementors)
+
+	out := graphql.NewFieldSet(fields)
+	var invalids uint32
+	for i, field := range fields {
+		switch field.Name {
+		case "__typename":
+			out.Values[i] = graphql.MarshalString("NotificationMarkAllAsReadResult")
+		case "success":
+			out.Values[i] = ec._NotificationMarkAllAsReadResult_success(ctx, field, obj)
+			if out.Values[i] == graphql.Null {
+				invalids++
+			}
+		default:
+			panic("unknown field " + strconv.Quote(field.Name))
+		}
+	}
+	out.Dispatch()
+	if invalids > 0 {
+		return graphql.Null
+	}
+	return out
+}
+
 var notifiedImplementors = []string{"Notified"}
 
 func (ec *executionContext) _Notified(ctx context.Context, sel ast.SelectionSet, obj *Notified) graphql.Marshaler {
@@ -27106,6 +27233,20 @@ func (ec *executionContext) marshalNNotificationFilter2githubᚗcomᚋjordanknot
 	return v
 }
 
+func (ec *executionContext) marshalNNotificationMarkAllAsReadResult2githubᚗcomᚋjordanknottᚋtaskcafeᚋinternalᚋgraphᚐNotificationMarkAllAsReadResult(ctx context.Context, sel ast.SelectionSet, v NotificationMarkAllAsReadResult) graphql.Marshaler {
+	return ec._NotificationMarkAllAsReadResult(ctx, sel, &v)
+}
+
+func (ec *executionContext) marshalNNotificationMarkAllAsReadResult2ᚖgithubᚗcomᚋjordanknottᚋtaskcafeᚋinternalᚋgraphᚐNotificationMarkAllAsReadResult(ctx context.Context, sel ast.SelectionSet, v *NotificationMarkAllAsReadResult) graphql.Marshaler {
+	if v == nil {
+		if !graphql.HasFieldError(ctx, graphql.GetFieldContext(ctx)) {
+			ec.Errorf(ctx, "must not be null")
+		}
+		return graphql.Null
+	}
+	return ec._NotificationMarkAllAsReadResult(ctx, sel, v)
+}
+
 func (ec *executionContext) unmarshalNNotificationToggleReadInput2githubᚗcomᚋjordanknottᚋtaskcafeᚋinternalᚋgraphᚐNotificationToggleReadInput(ctx context.Context, v interface{}) (NotificationToggleReadInput, error) {
 	res, err := ec.unmarshalInputNotificationToggleReadInput(ctx, v)
 	return res, graphql.ErrorOnPath(ctx, err)

internal/graph/graph.go

@@ -17,27 +17,38 @@ import (
 	"github.com/99designs/gqlgen/graphql/handler/lru"
 	"github.com/99designs/gqlgen/graphql/handler/transport"
 	"github.com/99designs/gqlgen/graphql/playground"
+	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
 	"github.com/jordanknott/taskcafe/internal/config"
 	"github.com/jordanknott/taskcafe/internal/db"
+	"github.com/jordanknott/taskcafe/internal/jobs"
 	"github.com/jordanknott/taskcafe/internal/logger"
 	"github.com/jordanknott/taskcafe/internal/utils"
 	log "github.com/sirupsen/logrus"
 	"github.com/vektah/gqlparser/v2/gqlerror"
 )
 
+type NotificationObservers struct {
+	Mu          sync.Mutex
+	Subscribers map[string]map[string]chan *Notified
+}
+
 // NewHandler returns a new graphql endpoint handler.
-func NewHandler(repo db.Repository, appConfig config.AppConfig) http.Handler {
+func NewHandler(repo db.Repository, appConfig config.AppConfig, jobQueue jobs.JobQueue, redisClient *redis.Client) http.Handler {
-	c := Config{
+	resolver := &Resolver{
-		Resolvers: &Resolver{
+		Repository: repo,
-			Repository: repo,
+		Redis:      redisClient,
-			AppConfig:  appConfig,
+		AppConfig:  appConfig,
-			Notifications: NotificationObservers{
+		Job:        jobQueue,
-				Mu:          sync.Mutex{},
+		Notifications: &NotificationObservers{
-				Subscribers: make(map[string]map[string]chan *Notified),
+			Mu:          sync.Mutex{},
-			},
+			Subscribers: make(map[string]map[string]chan *Notified),
 		},
 	}
+	resolver.SubscribeRedis()
+	c := Config{
+		Resolvers: resolver,
+	}
 	c.Directives.HasRole = func(ctx context.Context, obj interface{}, next graphql.Resolver, roles []RoleLevel, level ActionLevel, typeArg ObjectType) (interface{}, error) {
 		userID, ok := GetUser(ctx)
 		if !ok {

internal/graph/models_gen.go

@@ -402,6 +402,10 @@ type NotificationData struct {
 	Value string `json:"value"`
 }
 
+type NotificationMarkAllAsReadResult struct {
+	Success bool `json:"success"`
+}
+
 type NotificationToggleReadInput struct {
 	NotifiedID uuid.UUID `json:"notifiedID"`
 }
@@ -749,6 +753,7 @@ const (
 	ActionTypeDueDateAdded           ActionType = "DUE_DATE_ADDED"
 	ActionTypeDueDateRemoved         ActionType = "DUE_DATE_REMOVED"
 	ActionTypeDueDateChanged         ActionType = "DUE_DATE_CHANGED"
+	ActionTypeDueDateReminder        ActionType = "DUE_DATE_REMINDER"
 	ActionTypeTaskAssigned           ActionType = "TASK_ASSIGNED"
 	ActionTypeTaskMoved              ActionType = "TASK_MOVED"
 	ActionTypeTaskArchived           ActionType = "TASK_ARCHIVED"
@@ -766,6 +771,7 @@ var AllActionType = []ActionType{
 	ActionTypeDueDateAdded,
 	ActionTypeDueDateRemoved,
 	ActionTypeDueDateChanged,
+	ActionTypeDueDateReminder,
 	ActionTypeTaskAssigned,
 	ActionTypeTaskMoved,
 	ActionTypeTaskArchived,
@@ -776,7 +782,7 @@ var AllActionType = []ActionType{
 
 func (e ActionType) IsValid() bool {
 	switch e {
-	case ActionTypeTeamAdded, ActionTypeTeamRemoved, ActionTypeProjectAdded, ActionTypeProjectRemoved, ActionTypeProjectArchived, ActionTypeDueDateAdded, ActionTypeDueDateRemoved, ActionTypeDueDateChanged, ActionTypeTaskAssigned, ActionTypeTaskMoved, ActionTypeTaskArchived, ActionTypeTaskAttachmentUploaded, ActionTypeCommentMentioned, ActionTypeCommentOther:
+	case ActionTypeTeamAdded, ActionTypeTeamRemoved, ActionTypeProjectAdded, ActionTypeProjectRemoved, ActionTypeProjectArchived, ActionTypeDueDateAdded, ActionTypeDueDateRemoved, ActionTypeDueDateChanged, ActionTypeDueDateReminder, ActionTypeTaskAssigned, ActionTypeTaskMoved, ActionTypeTaskArchived, ActionTypeTaskAttachmentUploaded, ActionTypeCommentMentioned, ActionTypeCommentOther:
 		return true
 	}
 	return false

internal/graph/notification.resolvers.go

@@ -70,6 +70,19 @@ func (r *mutationResolver) NotificationToggleRead(ctx context.Context, input Not
 	}, nil
 }
 
+func (r *mutationResolver) NotificationMarkAllRead(ctx context.Context) (*NotificationMarkAllAsReadResult, error) {
+	userID, ok := GetUserID(ctx)
+	if !ok {
+		return &NotificationMarkAllAsReadResult{}, errors.New("invalid user ID")
+	}
+	now := time.Now().UTC()
+	err := r.Repository.MarkAllNotificationsRead(ctx, db.MarkAllNotificationsReadParams{UserID: userID, ReadAt: sql.NullTime{Valid: true, Time: now}})
+	if err != nil {
+		return &NotificationMarkAllAsReadResult{}, err
+	}
+	return &NotificationMarkAllAsReadResult{Success: false}, nil
+}
+
 func (r *notificationResolver) ID(ctx context.Context, obj *db.Notification) (uuid.UUID, error) {
 	return obj.NotificationID, nil
 }

internal/graph/resolver.go

@@ -4,20 +4,67 @@
 package graph
 
 import (
-	"sync"
+	"context"
+	"encoding/json"
 
+	"github.com/go-redis/redis/v8"
+	"github.com/google/uuid"
 	"github.com/jordanknott/taskcafe/internal/config"
 	"github.com/jordanknott/taskcafe/internal/db"
+	"github.com/jordanknott/taskcafe/internal/jobs"
+	"github.com/jordanknott/taskcafe/internal/utils"
+	log "github.com/sirupsen/logrus"
 )
 
-type NotificationObservers struct {
-	Subscribers map[string]map[string]chan *Notified
-	Mu          sync.Mutex
-}
-
 // Resolver handles resolving GraphQL queries & mutations
 type Resolver struct {
 	Repository    db.Repository
 	AppConfig     config.AppConfig
-	Notifications NotificationObservers
+	Notifications *NotificationObservers
+	Job           jobs.JobQueue
+	Redis         *redis.Client
+}
+
+func (r Resolver) SubscribeRedis() {
+	ctx := context.Background()
+	go func() {
+		subscriber := r.Redis.Subscribe(ctx, "notification-created")
+		log.Info("Stream starting...")
+		for {
+
+			msg, err := subscriber.ReceiveMessage(ctx)
+			if err != nil {
+				log.WithError(err).Error("while receiving message")
+				panic(err)
+			}
+			var message utils.NotificationCreatedMessage
+
+			if err := json.Unmarshal([]byte(msg.Payload), &message); err != nil {
+				log.WithError(err).Error("while unmarshalling notifiction created message")
+				panic(err)
+			}
+			log.WithField("notID", message.NotifiedID).Info("received notification message")
+
+			notified, err := r.Repository.GetNotifiedByIDNoExtra(ctx, uuid.MustParse(message.NotifiedID))
+			if err != nil {
+				log.WithError(err).Error("while getting notified by id")
+				panic(err)
+			}
+			notification, err := r.Repository.GetNotificationByID(ctx, uuid.MustParse(message.NotificationID))
+			if err != nil {
+				log.WithError(err).Error("while getting notified by id")
+				panic(err)
+			}
+			for _, observers := range r.Notifications.Subscribers {
+				for _, ochan := range observers {
+					ochan <- &Notified{
+						ID:           notified.NotifiedID,
+						Read:         notified.Read,
+						ReadAt:       &notified.ReadAt.Time,
+						Notification: &notification,
+					}
+				}
+			}
+		}
+	}()
 }

internal/graph/schema/notification.gql

@@ -10,6 +10,10 @@ extend type Query {
 
 extend type Mutation {
   notificationToggleRead(input: NotificationToggleReadInput!): Notified!
+  notificationMarkAllRead: NotificationMarkAllAsReadResult!
+}
+type NotificationMarkAllAsReadResult {
+  success: Boolean!
 }
 
 type HasUnreadNotificationsResult {
@@ -45,6 +49,7 @@ enum ActionType {
   DUE_DATE_ADDED
   DUE_DATE_REMOVED
   DUE_DATE_CHANGED
+  DUE_DATE_REMINDER
   TASK_ASSIGNED
   TASK_MOVED
   TASK_ARCHIVED

internal/graph/schema/notification/notification.gql

@@ -10,6 +10,10 @@ extend type Query {
 
 extend type Mutation {
   notificationToggleRead(input: NotificationToggleReadInput!): Notified!
+  notificationMarkAllRead: NotificationMarkAllAsReadResult!
+}
+type NotificationMarkAllAsReadResult {
+  success: Boolean!
 }
 
 type HasUnreadNotificationsResult {
@@ -45,6 +49,7 @@ enum ActionType {
   DUE_DATE_ADDED
   DUE_DATE_REMOVED
   DUE_DATE_CHANGED
+  DUE_DATE_REMINDER
   TASK_ASSIGNED
   TASK_MOVED
   TASK_ARCHIVED

internal/graph/task.resolvers.go

@@ -11,7 +11,10 @@ import (
 	"strconv"
 	"time"
 
+	mTasks "github.com/RichardKnop/machinery/v1/tasks"
+	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
+	"github.com/jinzhu/now"
 	"github.com/jordanknott/taskcafe/internal/db"
 	"github.com/jordanknott/taskcafe/internal/logger"
 	log "github.com/sirupsen/logrus"
@@ -539,6 +542,42 @@ func (r *mutationResolver) UpdateTaskDueDate(ctx context.Context, input UpdateTa
 			DueDate: dueDate,
 			HasTime: input.HasTime,
 		})
+		reminders, err := r.Repository.GetDueDateRemindersForTaskID(ctx, input.TaskID)
+		if err != nil {
+			log.WithError(err).Error("error while getting due date reminders for task ID")
+			return &db.Task{}, err
+		}
+		if input.DueDate != nil {
+			for _, rem := range reminders {
+				remindAt := now.With(*input.DueDate).BeginningOfDay()
+				if input.HasTime {
+					remindAt = *input.DueDate
+				}
+				switch rem.Duration {
+				case "MINUTE":
+					remindAt = remindAt.Add(time.Duration(-rem.Period) * time.Minute)
+					break
+				case "HOUR":
+					remindAt = remindAt.Add(time.Duration(-rem.Period) * time.Hour)
+					break
+				case "DAY":
+					remindAt = remindAt.AddDate(0, 0, int(-rem.Period))
+					break
+				case "WEEK":
+					remindAt = remindAt.AddDate(0, 0, 7*int(-rem.Period))
+					break
+				}
+
+				_, err := r.Repository.UpdateDueDateReminderRemindAt(ctx, db.UpdateDueDateReminderRemindAtParams{
+					DueDateReminderID: rem.DueDateReminderID,
+					RemindAt:          remindAt,
+				})
+				if err != nil {
+					log.WithError(err).Error("error while updating due date reminder remind at")
+					return &db.Task{}, err
+				}
+			}
+		}
 		createdAt := time.Now().UTC()
 		d, _ := json.Marshal(data)
 		if !isSame {
@@ -686,12 +725,55 @@ func (r *mutationResolver) UnassignTask(ctx context.Context, input *UnassignTask
 
 func (r *mutationResolver) CreateTaskDueDateNotifications(ctx context.Context, input []CreateTaskDueDateNotification) (*CreateTaskDueDateNotificationsResult, error) {
 	reminders := []DueDateNotification{}
+	if len(input) == 0 {
+		return &CreateTaskDueDateNotificationsResult{}, nil
+	}
+	task, err := r.Repository.GetTaskByID(ctx, input[0].TaskID)
+	if err != nil {
+		log.WithError(err).Error("error while getting task by id")
+		return &CreateTaskDueDateNotificationsResult{}, nil
+	}
 	for _, in := range input {
+		remindAt := now.With(task.DueDate.Time).BeginningOfDay()
+		if task.HasTime {
+			remindAt = task.DueDate.Time
+		}
+		switch in.Duration {
+		case "MINUTE":
+			remindAt = remindAt.Add(time.Duration(-in.Period) * time.Minute)
+			break
+		case "HOUR":
+			remindAt = remindAt.Add(time.Duration(-in.Period) * time.Hour)
+			break
+		case "DAY":
+			remindAt = remindAt.AddDate(0, 0, int(-in.Period))
+			break
+		case "WEEK":
+			remindAt = remindAt.AddDate(0, 0, 7*int(-in.Period))
+			break
+		}
+
+		log.Info("task not found, sending task")
+
 		n, err := r.Repository.CreateDueDateReminder(ctx, db.CreateDueDateReminderParams{
 			TaskID:   in.TaskID,
 			Period:   int32(in.Period),
 			Duration: in.Duration.String(),
+			RemindAt: remindAt,
 		})
+		signature := &mTasks.Signature{
+			UUID: "due_date_reminder_" + n.DueDateReminderID.String(),
+			Name: "dueDateNotification",
+			ETA:  &remindAt,
+			Args: []mTasks.Arg{{
+				Type:  "string",
+				Value: n.DueDateReminderID.String(),
+			}, {
+				Type:  "string",
+				Value: in.TaskID.String(),
+			}},
+		}
+		r.Job.Server.SendTask(signature)
 		if err != nil {
 			return &CreateTaskDueDateNotificationsResult{}, err
 		}
@@ -713,15 +795,71 @@ func (r *mutationResolver) CreateTaskDueDateNotifications(ctx context.Context, i
 
 func (r *mutationResolver) UpdateTaskDueDateNotifications(ctx context.Context, input []UpdateTaskDueDateNotification) (*UpdateTaskDueDateNotificationsResult, error) {
 	reminders := []DueDateNotification{}
+	if len(input) == 0 {
+		return &UpdateTaskDueDateNotificationsResult{}, nil
+	}
 	for _, in := range input {
+		task, err := r.Repository.GetTaskForDueDateReminder(ctx, in.ID)
+		if err != nil {
+			log.WithError(err).Error("error while getting task by id")
+			return &UpdateTaskDueDateNotificationsResult{}, nil
+		}
+		current, err := r.Repository.GetDueDateReminderByID(ctx, in.ID)
+		if err != nil {
+			log.WithError(err).Error("error while getting task by id")
+			return &UpdateTaskDueDateNotificationsResult{}, nil
+		}
+
+		remindAt := now.With(task.DueDate.Time).BeginningOfDay()
+		if task.HasTime {
+			remindAt = task.DueDate.Time
+		}
+		switch in.Duration {
+		case "MINUTE":
+			remindAt = remindAt.Add(time.Duration(-in.Period) * time.Minute)
+			break
+		case "HOUR":
+			remindAt = remindAt.Add(time.Duration(-in.Period) * time.Hour)
+			break
+		case "DAY":
+			remindAt = remindAt.AddDate(0, 0, int(-in.Period))
+			break
+		case "WEEK":
+			remindAt = remindAt.AddDate(0, 0, 7*int(-in.Period))
+			break
+		}
+
 		n, err := r.Repository.UpdateDueDateReminder(ctx, db.UpdateDueDateReminderParams{
 			DueDateReminderID: in.ID,
 			Period:            int32(in.Period),
 			Duration:          in.Duration.String(),
+			RemindAt:          remindAt,
 		})
 		if err != nil {
 			return &UpdateTaskDueDateNotificationsResult{}, err
 		}
+		etaNano := strconv.FormatInt(current.RemindAt.UnixNano(), 10)
+		result, err := r.Redis.ZRangeByScore(ctx, "delayed_tasks", &redis.ZRangeBy{Max: etaNano, Min: etaNano}).Result()
+		if err != nil {
+			log.WithError(err).Error("error while getting due date reminder")
+		}
+		log.WithField("result", result).Info("result raw")
+		if len(result) != 0 {
+			r.Redis.ZRem(ctx, "delayed_tasks", result)
+		}
+		signature := &mTasks.Signature{
+			UUID: "due_date_reminder_" + n.DueDateReminderID.String(),
+			Name: "dueDateNotification",
+			ETA:  &remindAt,
+			Args: []mTasks.Arg{{
+				Type:  "string",
+				Value: n.DueDateReminderID.String(),
+			}, {
+				Type:  "string",
+				Value: task.TaskID.String(),
+			}},
+		}
+		r.Job.Server.SendTask(signature)
 		duration := DueDateNotificationDuration(n.Duration)
 		if !duration.IsValid() {
 			log.WithField("duration", n.Duration).Error("invalid duration found")
@@ -741,11 +879,21 @@ func (r *mutationResolver) UpdateTaskDueDateNotifications(ctx context.Context, i
 func (r *mutationResolver) DeleteTaskDueDateNotifications(ctx context.Context, input []DeleteTaskDueDateNotification) (*DeleteTaskDueDateNotificationsResult, error) {
 	ids := []uuid.UUID{}
 	for _, n := range input {
-		err := r.Repository.DeleteDueDateReminder(ctx, n.ID)
+		reminder, err := r.Repository.GetDueDateReminderByID(ctx, n.ID)
+		err = r.Repository.DeleteDueDateReminder(ctx, n.ID)
 		if err != nil {
 			log.WithError(err).Error("error while deleting task due date notification")
 			return &DeleteTaskDueDateNotificationsResult{}, err
 		}
+		etaNano := strconv.FormatInt(reminder.RemindAt.UnixNano(), 10)
+		result, err := r.Redis.ZRangeByScore(ctx, "delayed_tasks", &redis.ZRangeBy{Max: etaNano, Min: etaNano}).Result()
+		if err != nil {
+			log.WithError(err).Error("error while getting due date reminder")
+		}
+		log.WithField("result", result).Info("result raw")
+		if len(result) != 0 {
+			r.Redis.ZRem(ctx, "delayed_tasks", result)
+		}
 		ids = append(ids, n.ID)
 	}
 	return &DeleteTaskDueDateNotificationsResult{Notifications: ids}, nil

internal/jobs/jobs.go

@@ -1,33 +1,172 @@
 package jobs
 
 import (
+	"context"
+	"encoding/json"
+	"strconv"
+	"time"
+
 	"github.com/RichardKnop/machinery/v1"
+	mTasks "github.com/RichardKnop/machinery/v1/tasks"
+	"github.com/go-redis/redis/v8"
+	"github.com/jinzhu/now"
+	log "github.com/sirupsen/logrus"
 
 	"github.com/google/uuid"
 	"github.com/jordanknott/taskcafe/internal/config"
 	"github.com/jordanknott/taskcafe/internal/db"
+	"github.com/jordanknott/taskcafe/internal/utils"
 )
 
-func RegisterTasks(server *machinery.Server, repo db.Repository) {
+type NotifiedData struct {
-	tasks := JobTasks{repo}
+	Data map[string]string
+}
+
+func RegisterTasks(server *machinery.Server, repo db.Repository, appConfig config.AppConfig, messageQueue *redis.Client) {
+	tasks := JobTasks{Repository: repo, Server: server, AppConfig: appConfig, MessageQueue: messageQueue}
 	server.RegisterTasks(map[string]interface{}{
-		"taskMemberWasAdded": tasks.TaskMemberWasAdded,
+		"dueDateNotification":          tasks.DueDateNotification,
+		"scheduleDueDateNotifications": tasks.ScheduleDueDateNotifications,
 	})
 }
 
 type JobTasks struct {
-	Repository db.Repository
+	AppConfig    config.AppConfig
+	Repository   db.Repository
+	Server       *machinery.Server
+	MessageQueue *redis.Client
 }
 
-func (t *JobTasks) TaskMemberWasAdded(taskID, notifierID, notifiedID string) (bool, error) {
+func (t *JobTasks) ScheduleDueDateNotifications() (bool, error) {
+	ctx := context.Background()
+	// tomorrow := now.With(time.Now().UTC().AddDate(0, 0, 1))
+	today := now.With(time.Now().UTC())
+	start := today.BeginningOfDay()
+	log.WithFields(log.Fields{
+		"start": start,
+	}).Info("fetching duration")
+	reminders, err := t.Repository.GetDueDateRemindersForDuration(ctx, start)
+
+	if err != nil {
+		log.WithError(err).Error("error while getting due date reminder")
+	}
+	for _, rem := range reminders {
+		log.WithField("id", rem.DueDateReminderID).Info("found reminder")
+		signature := &mTasks.Signature{
+			UUID: "due_date_reminder_" + rem.DueDateReminderID.String(),
+			Name: "dueDateNotification",
+			ETA:  &rem.RemindAt,
+			Args: []mTasks.Arg{{
+				Type:  "string",
+				Value: rem.DueDateReminderID.String(),
+			}, {
+				Type:  "string",
+				Value: rem.TaskID.String(),
+			}},
+		}
+		log.WithField("nanoTime", signature.ETA.UnixNano()).Info("rem time")
+		etaNano := strconv.FormatInt(signature.ETA.UnixNano(), 10)
+		result, err := t.MessageQueue.ZRangeByScore(ctx, "delayed_tasks", &redis.ZRangeBy{Max: etaNano, Min: etaNano}).Result()
+		if err != nil {
+			log.WithError(err).Error("error while getting due date reminder")
+		}
+		log.WithField("result", result).Info("result raw")
+		if len(result) == 0 {
+			log.Info("task not found, sending task")
+			t.Server.SendTask(signature)
+		}
+	}
+	return true, nil
+}
+
+func (t *JobTasks) DueDateNotification(dueDateIDEncoded string, taskIDEncoded string) (bool, error) {
+	ctx := context.Background()
+	dueDateID, err := uuid.Parse(dueDateIDEncoded)
+	if err != nil {
+		log.WithError(err).Error("while parsing task ID")
+		return false, err
+	}
+	taskID, err := uuid.Parse(taskIDEncoded)
+	if err != nil {
+		log.WithError(err).Error("while parsing task ID")
+		return false, err
+	}
+	dueAt, err := t.Repository.GetDueDateReminderByID(ctx, dueDateID)
+	if err != nil {
+		log.WithError(err).Error("while getting task by id")
+		return false, err
+	}
+	task, err := t.Repository.GetTaskByID(ctx, taskID)
+	if err != nil {
+		log.WithError(err).Error("while getting task by id")
+		return false, err
+	}
+	projectInfo, err := t.Repository.GetProjectInfoForTask(ctx, taskID)
+	if err != nil {
+		log.WithError(err).Error("error while getting project info for task")
+		return false, err
+	}
+	data := map[string]string{
+		"TaskID":      task.ShortID,
+		"TaskName":    task.Name,
+		"ProjectID":   projectInfo.ProjectShortID,
+		"ProjectName": projectInfo.Name,
+		"DueAt":       dueAt.RemindAt.String(),
+	}
+
+	now := time.Now().UTC()
+	raw, err := json.Marshal(NotifiedData{Data: data})
+	if err != nil {
+		log.WithError(err).Error("error while marshal json data for notification")
+		return false, err
+	}
+	n, err := t.Repository.CreateNotification(ctx, db.CreateNotificationParams{
+		CausedBy:   uuid.UUID{},
+		ActionType: "DUE_DATE_REMINDER",
+		CreatedOn:  now,
+		Data:       json.RawMessage(raw),
+	})
+	if err != nil {
+		log.WithError(err).Error("error while creating notification")
+		return false, err
+	}
+	watchers, err := t.Repository.GetTaskWatchersForTask(ctx, taskID)
+	if err != nil {
+		log.WithError(err).Error("while getting watchers")
+		return false, err
+	}
+
+	for _, watcher := range watchers {
+		notified, err := t.Repository.CreateNotificationNotifed(ctx, db.CreateNotificationNotifedParams{
+			UserID:         watcher.UserID,
+			NotificationID: n.NotificationID,
+		})
+		if err != nil {
+			log.WithError(err).Error("error while creating notification notified object")
+			return false, err
+		}
+		payload, err := json.Marshal(utils.NotificationCreatedMessage{
+			NotifiedID:     notified.NotifiedID.String(),
+			NotificationID: n.NotificationID.String(),
+		})
+		if err != nil {
+			panic(err)
+		}
+
+		if err := t.MessageQueue.Publish(context.Background(), "notification-created", payload).Err(); err != nil {
+			panic(err)
+		}
+	}
+
 	return true, nil
 }
 
 type JobQueue struct {
-	AppConfig config.AppConfig
+	AppConfig  config.AppConfig
-	Server    *machinery.Server
+	Repository db.Repository
+	Server     *machinery.Server
 }
 
-func (q *JobQueue) TaskMemberWasAdded(taskID, notifier, notified uuid.UUID) error {
+func (q *JobQueue) DueDateNotification(notificationId uuid.UUID) error {
 	return nil
 }

internal/route/route.go

@@ -8,6 +8,7 @@ import (
 	"github.com/go-chi/chi"
 	"github.com/go-chi/chi/middleware"
 	"github.com/go-chi/cors"
+	"github.com/go-redis/redis/v8"
 	"github.com/jmoiron/sqlx"
 	log "github.com/sirupsen/logrus"
 
@@ -18,6 +19,7 @@ import (
 	"github.com/jordanknott/taskcafe/internal/db"
 	"github.com/jordanknott/taskcafe/internal/frontend"
 	"github.com/jordanknott/taskcafe/internal/graph"
+	"github.com/jordanknott/taskcafe/internal/jobs"
 	"github.com/jordanknott/taskcafe/internal/logger"
 )
 
@@ -67,7 +69,7 @@ type TaskcafeHandler struct {
 }
 
 // NewRouter creates a new router for chi
-func NewRouter(dbConnection *sqlx.DB, job *machinery.Server, appConfig config.AppConfig) (chi.Router, error) {
+func NewRouter(dbConnection *sqlx.DB, redisClient *redis.Client, jobServer *machinery.Server, appConfig config.AppConfig) (chi.Router, error) {
 	formatter := new(log.TextFormatter)
 	formatter.TimestampFormat = "02-01-2006 15:04:05"
 	formatter.FullTimestamp = true
@@ -107,10 +109,15 @@ func NewRouter(dbConnection *sqlx.DB, job *machinery.Server, appConfig config.Ap
 		mux.Post("/logger", taskcafeHandler.HandleClientLog)
 	})
 	auth := AuthenticationMiddleware{*repository}
+	jobQueue := jobs.JobQueue{
+		Repository: *repository,
+		AppConfig:  appConfig,
+		Server:     jobServer,
+	}
 	r.Group(func(mux chi.Router) {
 		mux.Use(auth.Middleware)
 		mux.Post("/users/me/avatar", taskcafeHandler.ProfileImageUpload)
-		mux.Mount("/graphql", graph.NewHandler(*repository, appConfig))
+		mux.Mount("/graphql", graph.NewHandler(*repository, appConfig, jobQueue, redisClient))
 	})
 
 	frontend := FrontendHandler{staticPath: "build", indexPath: "index.html"}

internal/utils/redis.go

@@ -0,0 +1,6 @@
+package utils
+
+type NotificationCreatedMessage struct {
+	NotifiedID     string
+	NotificationID string
+}

migrations/0071_add-task_due_date_notification-at-col.up.sql

@@ -0,0 +1 @@
+ALTER TABLE task_due_date_reminder ADD COLUMN remind_at timestamptz NOT NULL DEFAULT NOW();

migrations/0072_remove-not-null-from-notification-caused_by.up.sql

@@ -0,0 +1,2 @@
+ALTER TABLE notification ALTER COLUMN caused_by DROP NOT NULL;
+  UPDATE notification SET caused_by = null WHERE caused_by = '00000000-0000-0000-0000-000000000000';