Merge pull request #2465 from omnivore-app/feature/rss

feat: rss

packages/api/package.json

@@ -83,6 +83,7 @@
     "pg": "^8.3.3",
     "postgrator": "^4.2.0",
     "private-ip": "^2.3.3",
+    "rss-parser": "^3.13.0",
     "sanitize-html": "^2.3.2",
     "search-query-parser": "^1.6.0",
     "snake-case": "^3.0.3",

packages/api/src/elastic/types.ts

@@ -162,6 +162,7 @@ export interface Page {
   listenedAt?: Date
   wordsCount?: number
   recommendations?: Recommendation[]
+  rssFeedUrl?: string
 }
 
 export interface SearchItem {

packages/api/src/entity/subscription.ts

@@ -5,15 +5,13 @@ import {
   JoinColumn,
   ManyToOne,
   PrimaryGeneratedColumn,
-  Unique,
   UpdateDateColumn,
 } from 'typeorm'
-import { User } from './user'
-import { SubscriptionStatus } from '../generated/graphql'
+import { SubscriptionStatus, SubscriptionType } from '../generated/graphql'
 import { NewsletterEmail } from './newsletter_email'
+import { User } from './user'
 
 @Entity({ name: 'subscriptions' })
-@Unique(['name', 'user'])
 export class Subscription {
   @PrimaryGeneratedColumn('uuid')
   id!: string
@@ -31,9 +29,9 @@ export class Subscription {
   })
   status!: SubscriptionStatus
 
-  @ManyToOne(() => NewsletterEmail)
+  @ManyToOne(() => NewsletterEmail, { nullable: true })
   @JoinColumn({ name: 'newsletter_email_id' })
-  newsletterEmail!: NewsletterEmail
+  newsletterEmail?: NewsletterEmail | null
 
   @Column('text', { nullable: true })
   description?: string
@@ -50,9 +48,20 @@ export class Subscription {
   @Column('text', { nullable: true })
   icon?: string
 
-  @CreateDateColumn()
+  @Column('enum', {
+    enum: SubscriptionType,
+  })
+  type!: SubscriptionType
+
+  @Column('integer', { default: 0 })
+  count!: number
+
+  @Column('timestamp', { nullable: true })
+  lastFetchedAt?: Date | null
+
+  @CreateDateColumn({ default: () => 'CURRENT_TIMESTAMP' })
   createdAt!: Date
 
-  @UpdateDateColumn()
+  @UpdateDateColumn({ default: () => 'CURRENT_TIMESTAMP' })
   updatedAt!: Date
 }

packages/api/src/generated/graphql.ts

@@ -1292,6 +1292,7 @@ export type Mutation = {
   updatePage: UpdatePageResult;
   updateReminder: UpdateReminderResult;
   updateSharedComment: UpdateSharedCommentResult;
+  updateSubscription: UpdateSubscriptionResult;
   updateUser: UpdateUserResult;
   updateUserProfile: UpdateUserProfileResult;
   uploadFileRequest: UploadFileRequestResult;
@@ -1574,12 +1575,13 @@ export type MutationSetWebhookArgs = {
 
 
 export type MutationSubscribeArgs = {
-  name: Scalars['String'];
+  input: SubscribeInput;
 };
 
 
 export type MutationUnsubscribeArgs = {
   name: Scalars['String'];
+  subscriptionId?: InputMaybe<Scalars['ID']>;
 };
 
 
@@ -1618,6 +1620,11 @@ export type MutationUpdateSharedCommentArgs = {
 };
 
 
+export type MutationUpdateSubscriptionArgs = {
+  input: UpdateSubscriptionInput;
+};
+
+
 export type MutationUpdateUserArgs = {
   input: UpdateUserInput;
 };
@@ -1869,6 +1876,7 @@ export type QuerySharedArticleArgs = {
 
 export type QuerySubscriptionsArgs = {
   sort?: InputMaybe<SortParams>;
+  type?: InputMaybe<SubscriptionType>;
 };
 
 
@@ -2249,6 +2257,7 @@ export type SavePageInput = {
   labels?: InputMaybe<Array<CreateLabelInput>>;
   originalContent: Scalars['String'];
   parseResult?: InputMaybe<ParseResult>;
+  rssFeedUrl?: InputMaybe<Scalars['String']>;
   source: Scalars['String'];
   state?: InputMaybe<ArticleSavingRequestStatus>;
   title?: InputMaybe<Scalars['String']>;
@@ -2697,6 +2706,12 @@ export enum SubscribeErrorCode {
   Unauthorized = 'UNAUTHORIZED'
 }
 
+export type SubscribeInput = {
+  name?: InputMaybe<Scalars['String']>;
+  subscriptionType?: InputMaybe<SubscriptionType>;
+  url?: InputMaybe<Scalars['String']>;
+};
+
 export type SubscribeResult = SubscribeError | SubscribeSuccess;
 
 export type SubscribeSuccess = {
@@ -2706,13 +2721,16 @@ export type SubscribeSuccess = {
 
 export type Subscription = {
   __typename?: 'Subscription';
+  count: Scalars['Int'];
   createdAt: Scalars['Date'];
   description?: Maybe<Scalars['String']>;
   icon?: Maybe<Scalars['String']>;
   id: Scalars['ID'];
+  lastFetchedAt?: Maybe<Scalars['Date']>;
   name: Scalars['String'];
-  newsletterEmail: Scalars['String'];
+  newsletterEmail?: Maybe<Scalars['String']>;
   status: SubscriptionStatus;
+  type: SubscriptionType;
   unsubscribeHttpUrl?: Maybe<Scalars['String']>;
   unsubscribeMailTo?: Maybe<Scalars['String']>;
   updatedAt: Scalars['Date'];
@@ -2725,6 +2743,11 @@ export enum SubscriptionStatus {
   Unsubscribed = 'UNSUBSCRIBED'
 }
 
+export enum SubscriptionType {
+  Newsletter = 'NEWSLETTER',
+  Rss = 'RSS'
+}
+
 export type SubscriptionsError = {
   __typename?: 'SubscriptionsError';
   errorCodes: Array<SubscriptionsErrorCode>;
@@ -2979,6 +3002,31 @@ export type UpdateSharedCommentSuccess = {
   sharedComment: Scalars['String'];
 };
 
+export type UpdateSubscriptionError = {
+  __typename?: 'UpdateSubscriptionError';
+  errorCodes: Array<UpdateSubscriptionErrorCode>;
+};
+
+export enum UpdateSubscriptionErrorCode {
+  BadRequest = 'BAD_REQUEST',
+  NotFound = 'NOT_FOUND',
+  Unauthorized = 'UNAUTHORIZED'
+}
+
+export type UpdateSubscriptionInput = {
+  description?: InputMaybe<Scalars['String']>;
+  id: Scalars['ID'];
+  lastFetchedAt?: InputMaybe<Scalars['Date']>;
+  name?: InputMaybe<Scalars['String']>;
+};
+
+export type UpdateSubscriptionResult = UpdateSubscriptionError | UpdateSubscriptionSuccess;
+
+export type UpdateSubscriptionSuccess = {
+  __typename?: 'UpdateSubscriptionSuccess';
+  subscription: Subscription;
+};
+
 export type UpdateUserError = {
   __typename?: 'UpdateUserError';
   errorCodes: Array<UpdateUserErrorCode>;
@@ -3687,10 +3735,12 @@ export type ResolversTypes = {
   String: ResolverTypeWrapper<Scalars['String']>;
   SubscribeError: ResolverTypeWrapper<SubscribeError>;
   SubscribeErrorCode: SubscribeErrorCode;
+  SubscribeInput: SubscribeInput;
   SubscribeResult: ResolversTypes['SubscribeError'] | ResolversTypes['SubscribeSuccess'];
   SubscribeSuccess: ResolverTypeWrapper<SubscribeSuccess>;
   Subscription: ResolverTypeWrapper<{}>;
   SubscriptionStatus: SubscriptionStatus;
+  SubscriptionType: SubscriptionType;
   SubscriptionsError: ResolverTypeWrapper<SubscriptionsError>;
   SubscriptionsErrorCode: SubscriptionsErrorCode;
   SubscriptionsResult: ResolversTypes['SubscriptionsError'] | ResolversTypes['SubscriptionsSuccess'];
@@ -3741,6 +3791,11 @@ export type ResolversTypes = {
   UpdateSharedCommentInput: UpdateSharedCommentInput;
   UpdateSharedCommentResult: ResolversTypes['UpdateSharedCommentError'] | ResolversTypes['UpdateSharedCommentSuccess'];
   UpdateSharedCommentSuccess: ResolverTypeWrapper<UpdateSharedCommentSuccess>;
+  UpdateSubscriptionError: ResolverTypeWrapper<UpdateSubscriptionError>;
+  UpdateSubscriptionErrorCode: UpdateSubscriptionErrorCode;
+  UpdateSubscriptionInput: UpdateSubscriptionInput;
+  UpdateSubscriptionResult: ResolversTypes['UpdateSubscriptionError'] | ResolversTypes['UpdateSubscriptionSuccess'];
+  UpdateSubscriptionSuccess: ResolverTypeWrapper<UpdateSubscriptionSuccess>;
   UpdateUserError: ResolverTypeWrapper<UpdateUserError>;
   UpdateUserErrorCode: UpdateUserErrorCode;
   UpdateUserInput: UpdateUserInput;
@@ -4093,6 +4148,7 @@ export type ResolversParentTypes = {
   SortParams: SortParams;
   String: Scalars['String'];
   SubscribeError: SubscribeError;
+  SubscribeInput: SubscribeInput;
   SubscribeResult: ResolversParentTypes['SubscribeError'] | ResolversParentTypes['SubscribeSuccess'];
   SubscribeSuccess: SubscribeSuccess;
   Subscription: {};
@@ -4135,6 +4191,10 @@ export type ResolversParentTypes = {
   UpdateSharedCommentInput: UpdateSharedCommentInput;
   UpdateSharedCommentResult: ResolversParentTypes['UpdateSharedCommentError'] | ResolversParentTypes['UpdateSharedCommentSuccess'];
   UpdateSharedCommentSuccess: UpdateSharedCommentSuccess;
+  UpdateSubscriptionError: UpdateSubscriptionError;
+  UpdateSubscriptionInput: UpdateSubscriptionInput;
+  UpdateSubscriptionResult: ResolversParentTypes['UpdateSubscriptionError'] | ResolversParentTypes['UpdateSubscriptionSuccess'];
+  UpdateSubscriptionSuccess: UpdateSubscriptionSuccess;
   UpdateUserError: UpdateUserError;
   UpdateUserInput: UpdateUserInput;
   UpdateUserProfileError: UpdateUserProfileError;
@@ -5114,7 +5174,7 @@ export type MutationResolvers<ContextType = ResolverContext, ParentType extends
   setShareHighlight?: Resolver<ResolversTypes['SetShareHighlightResult'], ParentType, ContextType, RequireFields<MutationSetShareHighlightArgs, 'input'>>;
   setUserPersonalization?: Resolver<ResolversTypes['SetUserPersonalizationResult'], ParentType, ContextType, RequireFields<MutationSetUserPersonalizationArgs, 'input'>>;
   setWebhook?: Resolver<ResolversTypes['SetWebhookResult'], ParentType, ContextType, RequireFields<MutationSetWebhookArgs, 'input'>>;
-  subscribe?: Resolver<ResolversTypes['SubscribeResult'], ParentType, ContextType, RequireFields<MutationSubscribeArgs, 'name'>>;
+  subscribe?: Resolver<ResolversTypes['SubscribeResult'], ParentType, ContextType, RequireFields<MutationSubscribeArgs, 'input'>>;
   unsubscribe?: Resolver<ResolversTypes['UnsubscribeResult'], ParentType, ContextType, RequireFields<MutationUnsubscribeArgs, 'name'>>;
   updateHighlight?: Resolver<ResolversTypes['UpdateHighlightResult'], ParentType, ContextType, RequireFields<MutationUpdateHighlightArgs, 'input'>>;
   updateHighlightReply?: Resolver<ResolversTypes['UpdateHighlightReplyResult'], ParentType, ContextType, RequireFields<MutationUpdateHighlightReplyArgs, 'input'>>;
@@ -5123,6 +5183,7 @@ export type MutationResolvers<ContextType = ResolverContext, ParentType extends
   updatePage?: Resolver<ResolversTypes['UpdatePageResult'], ParentType, ContextType, RequireFields<MutationUpdatePageArgs, 'input'>>;
   updateReminder?: Resolver<ResolversTypes['UpdateReminderResult'], ParentType, ContextType, RequireFields<MutationUpdateReminderArgs, 'input'>>;
   updateSharedComment?: Resolver<ResolversTypes['UpdateSharedCommentResult'], ParentType, ContextType, RequireFields<MutationUpdateSharedCommentArgs, 'input'>>;
+  updateSubscription?: Resolver<ResolversTypes['UpdateSubscriptionResult'], ParentType, ContextType, RequireFields<MutationUpdateSubscriptionArgs, 'input'>>;
   updateUser?: Resolver<ResolversTypes['UpdateUserResult'], ParentType, ContextType, RequireFields<MutationUpdateUserArgs, 'input'>>;
   updateUserProfile?: Resolver<ResolversTypes['UpdateUserProfileResult'], ParentType, ContextType, RequireFields<MutationUpdateUserProfileArgs, 'input'>>;
   uploadFileRequest?: Resolver<ResolversTypes['UploadFileRequestResult'], ParentType, ContextType, RequireFields<MutationUploadFileRequestArgs, 'input'>>;
@@ -5745,13 +5806,16 @@ export type SubscribeSuccessResolvers<ContextType = ResolverContext, ParentType
 };
 
 export type SubscriptionResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['Subscription'] = ResolversParentTypes['Subscription']> = {
+  count?: SubscriptionResolver<ResolversTypes['Int'], "count", ParentType, ContextType>;
   createdAt?: SubscriptionResolver<ResolversTypes['Date'], "createdAt", ParentType, ContextType>;
   description?: SubscriptionResolver<Maybe<ResolversTypes['String']>, "description", ParentType, ContextType>;
   icon?: SubscriptionResolver<Maybe<ResolversTypes['String']>, "icon", ParentType, ContextType>;
   id?: SubscriptionResolver<ResolversTypes['ID'], "id", ParentType, ContextType>;
+  lastFetchedAt?: SubscriptionResolver<Maybe<ResolversTypes['Date']>, "lastFetchedAt", ParentType, ContextType>;
   name?: SubscriptionResolver<ResolversTypes['String'], "name", ParentType, ContextType>;
-  newsletterEmail?: SubscriptionResolver<ResolversTypes['String'], "newsletterEmail", ParentType, ContextType>;
+  newsletterEmail?: SubscriptionResolver<Maybe<ResolversTypes['String']>, "newsletterEmail", ParentType, ContextType>;
   status?: SubscriptionResolver<ResolversTypes['SubscriptionStatus'], "status", ParentType, ContextType>;
+  type?: SubscriptionResolver<ResolversTypes['SubscriptionType'], "type", ParentType, ContextType>;
   unsubscribeHttpUrl?: SubscriptionResolver<Maybe<ResolversTypes['String']>, "unsubscribeHttpUrl", ParentType, ContextType>;
   unsubscribeMailTo?: SubscriptionResolver<Maybe<ResolversTypes['String']>, "unsubscribeMailTo", ParentType, ContextType>;
   updatedAt?: SubscriptionResolver<ResolversTypes['Date'], "updatedAt", ParentType, ContextType>;
@@ -5916,6 +5980,20 @@ export type UpdateSharedCommentSuccessResolvers<ContextType = ResolverContext, P
   __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
 };
 
+export type UpdateSubscriptionErrorResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['UpdateSubscriptionError'] = ResolversParentTypes['UpdateSubscriptionError']> = {
+  errorCodes?: Resolver<Array<ResolversTypes['UpdateSubscriptionErrorCode']>, ParentType, ContextType>;
+  __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
+};
+
+export type UpdateSubscriptionResultResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['UpdateSubscriptionResult'] = ResolversParentTypes['UpdateSubscriptionResult']> = {
+  __resolveType: TypeResolveFn<'UpdateSubscriptionError' | 'UpdateSubscriptionSuccess', ParentType, ContextType>;
+};
+
+export type UpdateSubscriptionSuccessResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['UpdateSubscriptionSuccess'] = ResolversParentTypes['UpdateSubscriptionSuccess']> = {
+  subscription?: Resolver<ResolversTypes['Subscription'], ParentType, ContextType>;
+  __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
+};
+
 export type UpdateUserErrorResolvers<ContextType = ResolverContext, ParentType extends ResolversParentTypes['UpdateUserError'] = ResolversParentTypes['UpdateUserError']> = {
   errorCodes?: Resolver<Array<ResolversTypes['UpdateUserErrorCode']>, ParentType, ContextType>;
   __isTypeOf?: IsTypeOfResolverFn<ParentType, ContextType>;
@@ -6383,6 +6461,9 @@ export type Resolvers<ContextType = ResolverContext> = {
   UpdateSharedCommentError?: UpdateSharedCommentErrorResolvers<ContextType>;
   UpdateSharedCommentResult?: UpdateSharedCommentResultResolvers<ContextType>;
   UpdateSharedCommentSuccess?: UpdateSharedCommentSuccessResolvers<ContextType>;
+  UpdateSubscriptionError?: UpdateSubscriptionErrorResolvers<ContextType>;
+  UpdateSubscriptionResult?: UpdateSubscriptionResultResolvers<ContextType>;
+  UpdateSubscriptionSuccess?: UpdateSubscriptionSuccessResolvers<ContextType>;
   UpdateUserError?: UpdateUserErrorResolvers<ContextType>;
   UpdateUserProfileError?: UpdateUserProfileErrorResolvers<ContextType>;
   UpdateUserProfileResult?: UpdateUserProfileResultResolvers<ContextType>;

packages/api/src/generated/schema.graphql

@@ -1151,8 +1151,8 @@ type Mutation {
   setShareHighlight(input: SetShareHighlightInput!): SetShareHighlightResult!
   setUserPersonalization(input: SetUserPersonalizationInput!): SetUserPersonalizationResult!
   setWebhook(input: SetWebhookInput!): SetWebhookResult!
-  subscribe(name: String!): SubscribeResult!
-  unsubscribe(name: String!): UnsubscribeResult!
+  subscribe(input: SubscribeInput!): SubscribeResult!
+  unsubscribe(name: String!, subscriptionId: ID): UnsubscribeResult!
   updateHighlight(input: UpdateHighlightInput!): UpdateHighlightResult!
   updateHighlightReply(input: UpdateHighlightReplyInput!): UpdateHighlightReplyResult!
   updateLabel(input: UpdateLabelInput!): UpdateLabelResult!
@@ -1160,6 +1160,7 @@ type Mutation {
   updatePage(input: UpdatePageInput!): UpdatePageResult!
   updateReminder(input: UpdateReminderInput!): UpdateReminderResult!
   updateSharedComment(input: UpdateSharedCommentInput!): UpdateSharedCommentResult!
+  updateSubscription(input: UpdateSubscriptionInput!): UpdateSubscriptionResult!
   updateUser(input: UpdateUserInput!): UpdateUserResult!
   updateUserProfile(input: UpdateUserProfileInput!): UpdateUserProfileResult!
   uploadFileRequest(input: UploadFileRequestInput!): UploadFileRequestResult!
@@ -1308,7 +1309,7 @@ type Query {
   search(after: String, first: Int, format: String, includeContent: Boolean, query: String): SearchResult!
   sendInstallInstructions: SendInstallInstructionsResult!
   sharedArticle(selectedHighlightId: String, slug: String!, username: String!): SharedArticleResult!
-  subscriptions(sort: SortParams): SubscriptionsResult!
+  subscriptions(sort: SortParams, type: SubscriptionType): SubscriptionsResult!
   typeaheadSearch(first: Int, query: String!): TypeaheadSearchResult!
   updatesSince(after: String, first: Int, since: Date!, sort: SortParams): UpdatesSinceResult!
   user(userId: ID, username: String): UserResult!
@@ -1636,6 +1637,7 @@ input SavePageInput {
   labels: [CreateLabelInput!]
   originalContent: String!
   parseResult: ParseResult
+  rssFeedUrl: String
   source: String!
   state: ArticleSavingRequestStatus
   title: String
@@ -2051,6 +2053,12 @@ enum SubscribeErrorCode {
   UNAUTHORIZED
 }
 
+input SubscribeInput {
+  name: String
+  subscriptionType: SubscriptionType
+  url: String
+}
+
 union SubscribeResult = SubscribeError | SubscribeSuccess
 
 type SubscribeSuccess {
@@ -2058,13 +2066,16 @@ type SubscribeSuccess {
 }
 
 type Subscription {
+  count: Int!
   createdAt: Date!
   description: String
   icon: String
   id: ID!
+  lastFetchedAt: Date
   name: String!
-  newsletterEmail: String!
+  newsletterEmail: String
   status: SubscriptionStatus!
+  type: SubscriptionType!
   unsubscribeHttpUrl: String
   unsubscribeMailTo: String
   updatedAt: Date!
@@ -2077,6 +2088,11 @@ enum SubscriptionStatus {
   UNSUBSCRIBED
 }
 
+enum SubscriptionType {
+  NEWSLETTER
+  RSS
+}
+
 type SubscriptionsError {
   errorCodes: [SubscriptionsErrorCode!]!
 }
@@ -2309,6 +2325,29 @@ type UpdateSharedCommentSuccess {
   sharedComment: String!
 }
 
+type UpdateSubscriptionError {
+  errorCodes: [UpdateSubscriptionErrorCode!]!
+}
+
+enum UpdateSubscriptionErrorCode {
+  BAD_REQUEST
+  NOT_FOUND
+  UNAUTHORIZED
+}
+
+input UpdateSubscriptionInput {
+  description: String
+  id: ID!
+  lastFetchedAt: Date
+  name: String
+}
+
+union UpdateSubscriptionResult = UpdateSubscriptionError | UpdateSubscriptionSuccess
+
+type UpdateSubscriptionSuccess {
+  subscription: Subscription!
+}
+
 type UpdateUserError {
   errorCodes: [UpdateUserErrorCode!]!
 }

packages/api/src/resolvers/function_resolvers.ts

@@ -5,6 +5,7 @@
 /* eslint-disable @typescript-eslint/explicit-module-boundary-types */
 import { getShareInfoForArticle } from '../datalayer/links/share_info'
 import { getPageByParam } from '../elastic/pages'
+import { Subscription } from '../entity/subscription'
 import {
   Article,
   ArticleHighlightsInput,
@@ -109,6 +110,7 @@ import {
   updateReminderResolver,
   updateSharedCommentResolver,
   updatesSinceResolver,
+  updateSubscriptionResolver,
   updateUserProfileResolver,
   updateUserResolver,
   uploadFileRequestResolver,
@@ -206,6 +208,7 @@ export const functionResolvers = {
     bulkAction: bulkActionResolver,
     importFromIntegration: importFromIntegrationResolver,
     setFavoriteArticle: setFavoriteArticleResolver,
+    updateSubscription: updateSubscriptionResolver,
   },
   Query: {
     me: getMeUserResolver,
@@ -573,6 +576,16 @@ export const functionResolvers = {
       return item.pageType || PageType.Unknown
     },
   },
+  Subscription: {
+    newsletterEmail(subscription: Subscription) {
+      return subscription.newsletterEmail?.address
+    },
+    icon(subscription: Subscription) {
+      return (
+        subscription.icon && createImageProxyUrl(subscription.icon, 128, 128)
+      )
+    },
+  },
   ...resultResolveTypeResolver('Login'),
   ...resultResolveTypeResolver('LogOut'),
   ...resultResolveTypeResolver('GoogleSignup'),
@@ -662,4 +675,5 @@ export const functionResolvers = {
   ...resultResolveTypeResolver('BulkAction'),
   ...resultResolveTypeResolver('ImportFromIntegration'),
   ...resultResolveTypeResolver('SetFavoriteArticle'),
+  ...resultResolveTypeResolver('UpdateSubscription'),
 }

packages/api/src/resolvers/subscriptions/index.ts

@@ -1,4 +1,4 @@
-import { ILike } from 'typeorm'
+import Parser from 'rss-parser'
 import { Subscription } from '../../entity/subscription'
 import { User } from '../../entity/user'
 import { getRepository } from '../../entity/utils'
@@ -6,6 +6,7 @@ import { env } from '../../env'
 import {
   MutationSubscribeArgs,
   MutationUnsubscribeArgs,
+  MutationUpdateSubscriptionArgs,
   QuerySubscriptionsArgs,
   SortBy,
   SortOrder,
@@ -16,20 +17,32 @@ import {
   SubscriptionsErrorCode,
   SubscriptionsSuccess,
   SubscriptionStatus,
+  SubscriptionType,
   UnsubscribeError,
   UnsubscribeErrorCode,
   UnsubscribeSuccess,
+  UpdateSubscriptionError,
+  UpdateSubscriptionErrorCode,
+  UpdateSubscriptionSuccess,
 } from '../../generated/graphql'
 import { getSubscribeHandler, unsubscribe } from '../../services/subscriptions'
+import { Merge } from '../../util'
 import { analytics } from '../../utils/analytics'
 import { authorized } from '../../utils/helpers'
-import { createImageProxyUrl } from '../../utils/imageproxy'
 
-export const subscriptionsResolver = authorized<
+type PartialSubscription = Omit<Subscription, 'newsletterEmail'>
+
+const parser = new Parser()
+
+export type SubscriptionsSuccessPartial = Merge<
   SubscriptionsSuccess,
+  { subscriptions: PartialSubscription[] }
+>
+export const subscriptionsResolver = authorized<
+  SubscriptionsSuccessPartial,
   SubscriptionsError,
   QuerySubscriptionsArgs
->(async (_obj, { sort }, { claims: { uid }, log }) => {
+>(async (_obj, { sort, type: subscriptionType }, { claims: { uid }, log }) => {
   log.info('subscriptionsResolver')
 
   analytics.track({
@@ -41,7 +54,8 @@ export const subscriptionsResolver = authorized<
   })
 
   try {
-    const sortBy = sort?.by === SortBy.UpdatedTime ? 'updatedAt' : 'createdAt'
+    const sortBy =
+      sort?.by === SortBy.UpdatedTime ? 'lastFetchedAt' : 'createdAt'
     const sortOrder = sort?.order === SortOrder.Ascending ? 'ASC' : 'DESC'
     const user = await getRepository(User).findOneBy({ id: uid })
     if (!user) {
@@ -52,20 +66,17 @@ export const subscriptionsResolver = authorized<
 
     const subscriptions = await getRepository(Subscription)
       .createQueryBuilder('subscription')
-      .innerJoinAndSelect('subscription.newsletterEmail', 'newsletterEmail')
+      .leftJoinAndSelect('subscription.newsletterEmail', 'newsletterEmail')
       .where({
         user: { id: uid },
         status: SubscriptionStatus.Active,
+        type: subscriptionType || SubscriptionType.Newsletter, // default to newsletter
       })
       .orderBy('subscription.' + sortBy, sortOrder)
       .getMany()
 
     return {
-      subscriptions: subscriptions.map((s) => ({
-        ...s,
-        icon: s.icon && createImageProxyUrl(s.icon, 128, 128),
-        newsletterEmail: s.newsletterEmail.address,
-      })),
+      subscriptions,
     }
   } catch (error) {
     log.error(error)
@@ -75,11 +86,15 @@ export const subscriptionsResolver = authorized<
   }
 })
 
-export const unsubscribeResolver = authorized<
+export type UnsubscribeSuccessPartial = Merge<
   UnsubscribeSuccess,
+  { subscription: PartialSubscription }
+>
+export const unsubscribeResolver = authorized<
+  UnsubscribeSuccessPartial,
   UnsubscribeError,
   MutationUnsubscribeArgs
->(async (_, { name }, { claims: { uid }, log }) => {
+>(async (_, { name, subscriptionId }, { claims: { uid }, log }) => {
   log.info('unsubscribeResolver')
 
   try {
@@ -90,13 +105,20 @@ export const unsubscribeResolver = authorized<
       }
     }
 
-    const subscription = await getRepository(Subscription)
+    const queryBuilder = getRepository(Subscription)
       .createQueryBuilder('subscription')
-      .innerJoinAndSelect('subscription.newsletterEmail', 'newsletterEmail')
+      .leftJoinAndSelect('subscription.newsletterEmail', 'newsletterEmail')
       .where({ user: { id: uid } })
-      .andWhere('LOWER(name) = LOWER(:name)', { name }) // case insensitive
-      .getOne()
 
+    if (subscriptionId) {
+      // if subscriptionId is provided, ignore name
+      queryBuilder.andWhere({ id: subscriptionId })
+    } else {
+      // if subscriptionId is not provided, use name for old clients
+      queryBuilder.andWhere({ name })
+    }
+
+    const subscription = await queryBuilder.getOne()
     if (!subscription) {
       return {
         errorCodes: [UnsubscribeErrorCode.NotFound],
@@ -110,8 +132,12 @@ export const unsubscribeResolver = authorized<
       }
     }
 
-    if (!subscription.unsubscribeMailTo && !subscription.unsubscribeHttpUrl) {
-      log.info('No unsubscribe method found')
+    if (
+      subscription.type === SubscriptionType.Newsletter &&
+      !subscription.unsubscribeMailTo &&
+      !subscription.unsubscribeHttpUrl
+    ) {
+      log.info('No unsubscribe method found for newsletter subscription')
     }
 
     await unsubscribe(subscription)
@@ -126,10 +152,7 @@ export const unsubscribeResolver = authorized<
     })
 
     return {
-      subscription: {
-        ...subscription,
-        newsletterEmail: subscription.newsletterEmail.address,
-      },
+      subscription,
     }
   } catch (error) {
     log.error('failed to unsubscribe', error)
@@ -139,11 +162,15 @@ export const unsubscribeResolver = authorized<
   }
 })
 
-export const subscribeResolver = authorized<
+export type SubscribeSuccessPartial = Merge<
   SubscribeSuccess,
+  { subscriptions: PartialSubscription[] }
+>
+export const subscribeResolver = authorized<
+  SubscribeSuccessPartial,
   SubscribeError,
   MutationSubscribeArgs
->(async (_, { name }, { claims: { uid }, log }) => {
+>(async (_, { input }, { claims: { uid }, log }) => {
   log.info('subscribeResolver')
 
   try {
@@ -154,10 +181,13 @@ export const subscribeResolver = authorized<
       }
     }
 
+    // find existing subscription
     const subscription = await getRepository(Subscription).findOneBy({
-      name: ILike(name),
+      url: input.url || undefined,
+      name: input.name || undefined,
       user: { id: uid },
       status: SubscriptionStatus.Active,
+      type: input.subscriptionType || SubscriptionType.Rss, // default to rss
     })
     if (subscription) {
       return {
@@ -165,34 +195,61 @@ export const subscribeResolver = authorized<
       }
     }
 
-    const subscribeHandler = getSubscribeHandler(name)
-    if (!subscribeHandler) {
-      return {
-        errorCodes: [SubscribeErrorCode.NotFound],
-      }
-    }
-
-    const newSubscriptions = await subscribeHandler.handleSubscribe(uid, name)
-    if (!newSubscriptions) {
-      return {
-        errorCodes: [SubscribeErrorCode.BadRequest],
-      }
-    }
-
     analytics.track({
       userId: uid,
       event: 'subscribed',
       properties: {
-        name,
+        ...input,
         env: env.server.apiEnv,
       },
     })
 
+    // create new newsletter subscription
+    if (input.name && input.subscriptionType === SubscriptionType.Newsletter) {
+      const subscribeHandler = getSubscribeHandler(input.name)
+      if (!subscribeHandler) {
+        return {
+          errorCodes: [SubscribeErrorCode.NotFound],
+        }
+      }
+
+      const newSubscriptions = await subscribeHandler.handleSubscribe(
+        uid,
+        input.name
+      )
+      if (!newSubscriptions) {
+        return {
+          errorCodes: [SubscribeErrorCode.BadRequest],
+        }
+      }
+
+      return {
+        subscriptions: newSubscriptions,
+      }
+    }
+
+    // create new rss subscription
+    if (input.url) {
+      // validate rss feed
+      const feed = await parser.parseURL(input.url)
+
+      const newSubscription = await getRepository(Subscription).save({
+        name: feed.title,
+        url: input.url,
+        user: { id: uid },
+        type: SubscriptionType.Rss,
+        description: feed.description,
+        icon: feed.image?.url,
+      })
+
+      return {
+        subscriptions: [newSubscription],
+      }
+    }
+
+    log.info('missing url or name')
     return {
-      subscriptions: newSubscriptions.map((s) => ({
-        ...s,
-        newsletterEmail: s.newsletterEmail.address,
-      })),
+      errorCodes: [SubscribeErrorCode.BadRequest],
     }
   } catch (error) {
     log.error('failed to subscribe', error)
@@ -201,3 +258,65 @@ export const subscribeResolver = authorized<
     }
   }
 })
+
+export type UpdateSubscriptionSuccessPartial = Merge<
+  UpdateSubscriptionSuccess,
+  { subscription: PartialSubscription }
+>
+export const updateSubscriptionResolver = authorized<
+  UpdateSubscriptionSuccessPartial,
+  UpdateSubscriptionError,
+  MutationUpdateSubscriptionArgs
+>(async (_, { input }, { claims: { uid }, log }) => {
+  log.info('updateSubscriptionResolver')
+
+  try {
+    analytics.track({
+      userId: uid,
+      event: 'update_subscription',
+      properties: {
+        ...input,
+        env: env.server.apiEnv,
+      },
+    })
+
+    const user = await getRepository(User).findOneBy({ id: uid })
+    if (!user) {
+      return {
+        errorCodes: [UpdateSubscriptionErrorCode.Unauthorized],
+      }
+    }
+
+    // find existing subscription
+    const subscription = await getRepository(Subscription).findOneBy({
+      id: input.id,
+      user: { id: uid },
+      status: SubscriptionStatus.Active,
+    })
+    if (!subscription) {
+      log.info('subscription not found')
+      return {
+        errorCodes: [UpdateSubscriptionErrorCode.NotFound],
+      }
+    }
+
+    // update subscription
+    const updatedSubscription = await getRepository(Subscription).save({
+      id: input.id,
+      name: input.name || undefined,
+      description: input.description || undefined,
+      lastFetchedAt: input.lastFetchedAt
+        ? new Date(input.lastFetchedAt)
+        : undefined,
+    })
+
+    return {
+      subscription: updatedSubscription,
+    }
+  } catch (error) {
+    log.error('failed to update subscription', error)
+    return {
+      errorCodes: [UpdateSubscriptionErrorCode.BadRequest],
+    }
+  }
+})

packages/api/src/routers/svc/rss_feed.ts

@@ -0,0 +1,53 @@
+/* eslint-disable @typescript-eslint/no-misused-promises */
+import express from 'express'
+import { readPushSubscription } from '../../datalayer/pubsub'
+import { Subscription } from '../../entity/subscription'
+import { getRepository } from '../../entity/utils'
+import { SubscriptionStatus, SubscriptionType } from '../../generated/graphql'
+import { enqueueRssFeedFetch } from '../../utils/createTask'
+
+export function rssFeedRouter() {
+  const router = express.Router()
+
+  router.post('/fetchAll', async (req, res) => {
+    console.log('fetch all rss feeds')
+
+    const { message: msgStr, expired } = readPushSubscription(req)
+    console.log('read pubsub message', msgStr, 'has expired', expired)
+
+    if (expired) {
+      console.log('discarding expired message')
+      return res.status(200).send('Expired')
+    }
+
+    try {
+      // get all active rss feed subscriptions
+      const subscriptions = await getRepository(Subscription).find({
+        select: ['id', 'url', 'user', 'lastFetchedAt'],
+        where: {
+          type: SubscriptionType.Rss,
+          status: SubscriptionStatus.Active,
+        },
+        relations: ['user'],
+      })
+
+      // create a cloud taks to fetch rss feed item for each subscription
+      await Promise.all(
+        subscriptions.map((subscription) => {
+          try {
+            return enqueueRssFeedFetch(subscription)
+          } catch (error) {
+            console.log('error creating rss feed fetch task', error)
+          }
+        })
+      )
+
+      res.send('OK')
+    } catch (error) {
+      console.log('error fetching rss feeds', error)
+      res.status(500).send('Internal Server Error')
+    }
+  })
+
+  return router
+}

packages/api/src/schema.ts

@@ -561,6 +561,7 @@ const schema = gql`
     parseResult: ParseResult
     state: ArticleSavingRequestStatus
     labels: [CreateLabelInput!]
+    rssFeedUrl: String
   }
 
   input SaveUrlInput {
@@ -1619,16 +1620,24 @@ const schema = gql`
     subscriptions: [Subscription!]!
   }
 
+  enum SubscriptionType {
+    RSS
+    NEWSLETTER
+  }
+
   type Subscription {
     id: ID!
     name: String!
-    newsletterEmail: String!
+    newsletterEmail: String
     url: String
     description: String
     status: SubscriptionStatus!
     unsubscribeMailTo: String
     unsubscribeHttpUrl: String
     icon: String
+    type: SubscriptionType!
+    count: Int!
+    lastFetchedAt: Date
     createdAt: Date!
     updatedAt: Date!
   }
@@ -2478,6 +2487,37 @@ const schema = gql`
     ALREADY_EXISTS
   }
 
+  input SubscribeInput {
+    url: String
+    name: String
+    subscriptionType: SubscriptionType
+  }
+
+  input UpdateSubscriptionInput {
+    id: ID!
+    name: String
+    description: String
+    lastFetchedAt: Date
+  }
+
+  union UpdateSubscriptionResult =
+      UpdateSubscriptionSuccess
+    | UpdateSubscriptionError
+
+  type UpdateSubscriptionSuccess {
+    subscription: Subscription!
+  }
+
+  type UpdateSubscriptionError {
+    errorCodes: [UpdateSubscriptionErrorCode!]!
+  }
+
+  enum UpdateSubscriptionErrorCode {
+    UNAUTHORIZED
+    BAD_REQUEST
+    NOT_FOUND
+  }
+
   # Mutations
   type Mutation {
     googleLogin(input: GoogleLoginInput!): LoginResult!
@@ -2539,8 +2579,8 @@ const schema = gql`
     deleteLabel(id: ID!): DeleteLabelResult!
     setLabels(input: SetLabelsInput!): SetLabelsResult!
     generateApiKey(input: GenerateApiKeyInput!): GenerateApiKeyResult!
-    unsubscribe(name: String!): UnsubscribeResult!
-    subscribe(name: String!): SubscribeResult!
+    unsubscribe(name: String!, subscriptionId: ID): UnsubscribeResult!
+    subscribe(input: SubscribeInput!): SubscribeResult!
     addPopularRead(name: String!): AddPopularReadResult!
     setWebhook(input: SetWebhookInput!): SetWebhookResult!
     deleteWebhook(id: ID!): DeleteWebhookResult!
@@ -2576,6 +2616,9 @@ const schema = gql`
     ): BulkActionResult!
     importFromIntegration(integrationId: ID!): ImportFromIntegrationResult!
     setFavoriteArticle(id: ID!): SetFavoriteArticleResult!
+    updateSubscription(
+      input: UpdateSubscriptionInput!
+    ): UpdateSubscriptionResult!
   }
 
   # FIXME: remove sort from feedArticles after all cached tabs are closed
@@ -2620,7 +2663,10 @@ const schema = gql`
       includeContent: Boolean
       format: String
     ): SearchResult!
-    subscriptions(sort: SortParams): SubscriptionsResult!
+    subscriptions(
+      sort: SortParams
+      type: SubscriptionType
+    ): SubscriptionsResult!
     sendInstallInstructions: SendInstallInstructionsResult!
     webhooks: WebhooksResult!
     webhook(id: ID!): WebhookResult!

packages/api/src/server.ts

@@ -43,6 +43,7 @@ import { integrationsServiceRouter } from './routers/svc/integrations'
 import { linkServiceRouter } from './routers/svc/links'
 import { newsletterServiceRouter } from './routers/svc/newsletters'
 import { remindersServiceRouter } from './routers/svc/reminders'
+import { rssFeedRouter } from './routers/svc/rss_feed'
 import { uploadServiceRouter } from './routers/svc/upload'
 import { webhooksServiceRouter } from './routers/svc/webhooks'
 import { textToSpeechRouter } from './routers/text_to_speech'
@@ -159,6 +160,7 @@ export const createApp = (): {
   app.use('/svc/pubsub/integrations', integrationsServiceRouter())
   app.use('/svc/reminders', remindersServiceRouter())
   app.use('/svc/email-attachment', emailAttachmentRouter())
+  app.use('/svc/rss-feed', rssFeedRouter())
 
   if (env.dev.isLocal) {
     app.use('/local/debug', localDebugRouter())

packages/api/src/services/save_page.ts

@@ -103,6 +103,7 @@ export const savePage = async (
     pageType: parseResult.pageType,
     originalHtml: parseResult.domContent,
     canonicalUrl: parseResult.canonicalUrl,
+    rssFeedUrl: input.rssFeedUrl,
   })
 
   // save state
@@ -221,6 +222,7 @@ export const parsedContentToPage = ({
   uploadFileHash,
   uploadFileId,
   saveTime,
+  rssFeedUrl,
 }: {
   url: string
   userId: string
@@ -236,6 +238,7 @@ export const parsedContentToPage = ({
   uploadFileHash?: string | null
   uploadFileId?: string | null
   saveTime?: Date
+  rssFeedUrl?: string | null
 }): Page => {
   return {
     id: pageId || '',
@@ -267,5 +270,6 @@ export const parsedContentToPage = ({
     language: parsedContent?.language ?? undefined,
     siteIcon: parsedContent?.siteIcon ?? undefined,
     wordsCount: wordsCount(parsedContent?.textContent || ''),
+    rssFeedUrl: rssFeedUrl || undefined,
   }
 }

packages/api/src/services/subscriptions.ts

@@ -2,7 +2,7 @@ import axios from 'axios'
 import { NewsletterEmail } from '../entity/newsletter_email'
 import { Subscription } from '../entity/subscription'
 import { getRepository } from '../entity/utils'
-import { SubscriptionStatus } from '../generated/graphql'
+import { SubscriptionStatus, SubscriptionType } from '../generated/graphql'
 import { sendEmail } from '../utils/sendEmail'
 import { createNewsletterEmail } from './newsletters'
 
@@ -13,6 +13,7 @@ interface SaveSubscriptionInput {
   unsubscribeMailTo?: string
   unsubscribeHttpUrl?: string
   icon?: string
+  from?: string
 }
 
 export const UNSUBSCRIBE_EMAIL_TEXT =
@@ -85,6 +86,7 @@ export const getSubscriptionByNameAndUserId = async (
   return getRepository(Subscription).findOneBy({
     name,
     user: { id: userId },
+    type: SubscriptionType.Newsletter,
   })
 }
 
@@ -96,42 +98,63 @@ export const saveSubscription = async ({
   unsubscribeHttpUrl,
   icon,
 }: SaveSubscriptionInput): Promise<string> => {
-  const result = await getRepository(Subscription).upsert(
-    {
-      name,
-      newsletterEmail: { id: newsletterEmail.id },
-      user: { id: userId },
-      unsubscribeHttpUrl,
-      unsubscribeMailTo,
-      icon,
-    },
-    ['name', 'user']
-  )
+  const subscriptionData = {
+    unsubscribeHttpUrl,
+    unsubscribeMailTo,
+    icon,
+    lastFetchedAt: new Date(),
+  }
 
-  return result.identifiers[0].id as string
+  const existingSubscription = await getSubscriptionByNameAndUserId(
+    name,
+    userId
+  )
+  if (existingSubscription) {
+    // update subscription if already exists
+    await getRepository(Subscription).update(
+      existingSubscription.id,
+      subscriptionData
+    )
+
+    return existingSubscription.id
+  }
+
+  const result = await getRepository(Subscription).save({
+    ...subscriptionData,
+    name,
+    newsletterEmail: { id: newsletterEmail.id },
+    user: { id: userId },
+    type: SubscriptionType.Newsletter,
+  })
+
+  return result.id
 }
 
 export const unsubscribe = async (subscription: Subscription) => {
-  let unsubscribed = false
-  if (subscription.unsubscribeMailTo) {
-    // unsubscribe by sending email
-    unsubscribed = await sendUnsubscribeEmail(
-      subscription.unsubscribeMailTo,
-      subscription.newsletterEmail.address
-    )
-  }
-  // TODO: find a good way to unsubscribe by url if email fails or not provided
-  // because it often requires clicking a button on the page to unsubscribe
+  // unsubscribe from newsletter
+  if (subscription.type === SubscriptionType.Newsletter) {
+    let unsubscribed = false
 
-  if (!unsubscribed) {
-    // update subscription status to unsubscribed if failed to unsubscribe
-    console.log('Failed to unsubscribe', subscription.id)
-    return getRepository(Subscription).update(subscription.id, {
-      status: SubscriptionStatus.Unsubscribed,
-    })
+    if (subscription.unsubscribeMailTo && subscription.newsletterEmail) {
+      // unsubscribe by sending email
+      unsubscribed = await sendUnsubscribeEmail(
+        subscription.unsubscribeMailTo,
+        subscription.newsletterEmail.address
+      )
+    }
+    // TODO: find a good way to unsubscribe by url if email fails or not provided
+    // because it often requires clicking a button on the page to unsubscribe
+
+    if (!unsubscribed) {
+      // update subscription status to unsubscribed if failed to unsubscribe
+      console.log('Failed to unsubscribe', subscription.id)
+      return getRepository(Subscription).update(subscription.id, {
+        status: SubscriptionStatus.Unsubscribed,
+      })
+    }
   }
 
-  // delete the subscription if successfully unsubscribed
+  // delete the subscription if successfully unsubscribed or it's an rss feed
   await getRepository(Subscription).delete(subscription.id)
 }
 

packages/api/src/util.ts

@@ -1,8 +1,8 @@
 /* eslint-disable @typescript-eslint/no-unsafe-return */
 /* eslint-disable @typescript-eslint/naming-convention */
 /* eslint-disable @typescript-eslint/no-explicit-any */
-import os from 'os'
 import * as dotenv from 'dotenv'
+import os from 'os'
 
 interface BackendEnv {
   pg: {
@@ -67,6 +67,7 @@ interface BackendEnv {
     textToSpeechTaskHandlerUrl: string
     recommendationTaskHandlerUrl: string
     thumbnailTaskHandlerUrl: string
+    rssFeedTaskHandlerUrl: string
   }
   fileUpload: {
     gcsUploadBucket: string
@@ -161,6 +162,7 @@ const nullableEnvVars = [
   'RECOMMENDATION_TASK_HANDLER_URL',
   'POCKET_CONSUMER_KEY',
   'THUMBNAIL_TASK_HANDLER_URL',
+  'RSS_FEED_TASK_HANDLER_URL',
 ] // Allow some vars to be null/empty
 
 /* If not in GAE and Prod/QA/Demo env (f.e. on localhost/dev env), allow following env vars to be null */
@@ -248,6 +250,7 @@ export function getEnv(): BackendEnv {
     textToSpeechTaskHandlerUrl: parse('TEXT_TO_SPEECH_TASK_HANDLER_URL'),
     recommendationTaskHandlerUrl: parse('RECOMMENDATION_TASK_HANDLER_URL'),
     thumbnailTaskHandlerUrl: parse('THUMBNAIL_TASK_HANDLER_URL'),
+    rssFeedTaskHandlerUrl: parse('RSS_FEED_TASK_HANDLER_URL'),
   }
   const imageProxy = {
     url: parse('IMAGE_PROXY_URL'),

packages/api/src/utils/createTask.ts

@@ -6,6 +6,7 @@ import { google } from '@google-cloud/tasks/build/protos/protos'
 import axios from 'axios'
 import { nanoid } from 'nanoid'
 import { Recommendation } from '../elastic/types'
+import { Subscription } from '../entity/subscription'
 import { env } from '../env'
 import {
   ArticleSavingRequestStatus,
@@ -559,4 +560,32 @@ export const enqueueThumbnailTask = async (
   return createdTasks[0].name
 }
 
+export const enqueueRssFeedFetch = async (
+  rssFeedSubscription: Subscription
+): Promise<string> => {
+  const { GOOGLE_CLOUD_PROJECT } = process.env
+  const payload = {
+    subscriptionId: rssFeedSubscription.id,
+    userId: rssFeedSubscription.user.id,
+    feedUrl: rssFeedSubscription.url,
+    lastFetchedAt: rssFeedSubscription.lastFetchedAt,
+  }
+
+  const createdTasks = await createHttpTaskWithToken({
+    project: GOOGLE_CLOUD_PROJECT,
+    queue: 'omnivore-rss-feed-queue',
+    payload,
+    taskHandlerUrl: env.queue.rssFeedTaskHandlerUrl,
+  })
+
+  if (!createdTasks || !createdTasks[0].name) {
+    logger.error(`Unable to get the name of the task`, {
+      payload,
+      createdTasks,
+    })
+    throw new CreateTaskError(`Unable to get the name of the task`)
+  }
+  return createdTasks[0].name
+}
+
 export default createHttpTaskWithToken

packages/api/test/db.ts

@@ -1,20 +1,20 @@
 import Postgrator from 'postgrator'
-import { User } from '../src/entity/user'
-import { Profile } from '../src/entity/profile'
-import { Page } from '../src/entity/page'
-import { Link } from '../src/entity/link'
-import { Reminder } from '../src/entity/reminder'
-import { NewsletterEmail } from '../src/entity/newsletter_email'
-import { UserDeviceToken } from '../src/entity/user_device_tokens'
-import { Label } from '../src/entity/label'
-import { Subscription } from '../src/entity/subscription'
-import { AppDataSource } from '../src/server'
-import { getRepository, setClaims } from '../src/entity/utils'
-import { createUser } from '../src/services/create_user'
-import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
-import { SubscriptionStatus } from '../src/generated/graphql'
-import { Integration } from '../src/entity/integration'
 import { FindOptionsWhere } from 'typeorm'
+import { SnakeNamingStrategy } from 'typeorm-naming-strategies'
+import { Integration } from '../src/entity/integration'
+import { Label } from '../src/entity/label'
+import { Link } from '../src/entity/link'
+import { NewsletterEmail } from '../src/entity/newsletter_email'
+import { Page } from '../src/entity/page'
+import { Profile } from '../src/entity/profile'
+import { Reminder } from '../src/entity/reminder'
+import { Subscription } from '../src/entity/subscription'
+import { User } from '../src/entity/user'
+import { UserDeviceToken } from '../src/entity/user_device_tokens'
+import { getRepository, setClaims } from '../src/entity/utils'
+import { SubscriptionStatus, SubscriptionType } from '../src/generated/graphql'
+import { AppDataSource } from '../src/server'
+import { createUser } from '../src/services/create_user'
 
 const runMigrations = async () => {
   const migrationDirectory = __dirname + '/../../db/migrations'
@@ -199,7 +199,8 @@ export const createTestSubscription = async (
   name: string,
   newsletterEmail?: NewsletterEmail,
   status = SubscriptionStatus.Active,
-  unsubscribeMailTo?: string
+  unsubscribeMailTo?: string,
+  subscriptionType = SubscriptionType.Newsletter
 ): Promise<Subscription> => {
   return getRepository(Subscription).save({
     user,
@@ -207,6 +208,8 @@ export const createTestSubscription = async (
     newsletterEmail,
     status,
     unsubscribeMailTo,
+    lastFetchedAt: new Date(),
+    type: subscriptionType,
   })
 }
 

packages/api/test/resolvers/subscriptions.test.ts

@@ -6,7 +6,10 @@ import { NewsletterEmail } from '../../src/entity/newsletter_email'
 import { Subscription } from '../../src/entity/subscription'
 import { User } from '../../src/entity/user'
 import { getRepository } from '../../src/entity/utils'
-import { SubscriptionStatus } from '../../src/generated/graphql'
+import {
+  SubscriptionStatus,
+  SubscriptionType,
+} from '../../src/generated/graphql'
 import { UNSUBSCRIBE_EMAIL_TEXT } from '../../src/services/subscriptions'
 import * as sendEmail from '../../src/utils/sendEmail'
 import { createTestSubscription, createTestUser, deleteTestUser } from '../db'
@@ -35,7 +38,7 @@ describe('Subscriptions API', () => {
       confirmationCode: 'test',
     })
 
-    //  create testing subscriptions
+    //  create testing newsletter subscriptions
     const sub1 = await createTestSubscription(user, 'sub_1', newsletterEmail)
     const sub2 = await createTestSubscription(user, 'sub_2', newsletterEmail)
     // create a unsubscribed subscription
@@ -45,8 +48,15 @@ describe('Subscriptions API', () => {
       newsletterEmail,
       SubscriptionStatus.Unsubscribed
     )
-    // create a subscription without a newsletter email
-    await createTestSubscription(user, 'sub_4')
+    // create an rss feed subscription
+    await createTestSubscription(
+      user,
+      'sub_4',
+      undefined,
+      SubscriptionStatus.Active,
+      undefined,
+      SubscriptionType.Rss
+    )
     subscriptions = [sub2, sub1]
   })
 
@@ -143,10 +153,7 @@ describe('Subscriptions API', () => {
         sinon.fake.resolves(true)
       )
 
-      const res = await graphqlRequest(
-        query(name.toUpperCase()),
-        authToken
-      ).expect(200)
+      const res = await graphqlRequest(query(name), authToken).expect(200)
 
       expect(res.body.data.unsubscribe.subscription).to.eql({
         id: subscription.id,

packages/db/elastic_migrations/index_settings.json

@@ -159,6 +159,10 @@
         "type": "keyword",
         "normalizer": "lowercase_normalizer"
       },
+      "rssFeedUrl": {
+        "type": "keyword",
+        "normalizer": "lowercase_normalizer"
+      },
       "state": {
         "type": "keyword"
       },

packages/db/migrations/0115.do.add_type_to_subscriptions.sql

@@ -0,0 +1,15 @@
+-- Type: DO
+-- Name: add_type_to_subscriptions
+-- Description: Add type, count and last_fetched_at fields to subscriptions table
+
+BEGIN;
+
+CREATE TYPE subscription_type AS ENUM ('NEWSLETTER', 'RSS');
+
+ALTER TABLE omnivore.subscriptions
+    ADD COLUMN "type" subscription_type NOT NULL DEFAULT 'NEWSLETTER',
+    ADD COLUMN count INTEGER NOT NULL DEFAULT 0,
+    ADD COLUMN last_fetched_at timestamptz,
+    DROP CONSTRAINT subscriptions_user_id_name_key; -- Drop unique constraint on user_id and name
+
+COMMIT;

packages/db/migrations/0115.undo.add_type_to_subscriptions.sql

@@ -0,0 +1,15 @@
+-- Type: UNDO
+-- Name: add_type_to_subscriptions
+-- Description: Add type, count and last_fetched_at fields to subscriptions table
+
+BEGIN;
+
+ALTER TABLE omnivore.subscriptions
+    ADD CONSTRAINT subscriptions_user_id_name_key UNIQUE (user_id, name),
+    DROP COLUMN last_fetched_at,
+    DROP COLUMN count,
+    DROP COLUMN "type";
+
+DROP TYPE subscription_type;
+
+COMMIT;

packages/rss-handler/.dockerignore

@@ -0,0 +1,5 @@
+node_modules
+build
+.env*
+Dockerfile
+.dockerignore

packages/rss-handler/.eslintignore

@@ -0,0 +1,2 @@
+node_modules/
+build/

packages/rss-handler/.eslintrc

@@ -0,0 +1,6 @@
+{
+  "extends": "../../.eslintrc",
+  "parserOptions": {
+    "project": "tsconfig.json"
+  }
+}

packages/rss-handler/.gcloudignore

@@ -0,0 +1,16 @@
+# This file specifies files that are *not* uploaded to Google Cloud Platform
+# using gcloud. It follows the same syntax as .gitignore, with the addition of
+# "#!include" directives (which insert the entries of the given .gitignore-style
+# file at that point).
+#
+# For more information, run:
+#   $ gcloud topic gcloudignore
+#
+.gcloudignore
+# If you would like to upload your .git directory, .gitignore file or files
+# from your .gitignore file, remove the corresponding line
+# below:
+.git
+.gitignore
+
+node_modules

packages/rss-handler/Dockerfile

@@ -0,0 +1,26 @@
+FROM node:14.18-alpine
+
+# Run everything after as non-privileged user.
+WORKDIR /app
+
+COPY package.json .
+COPY yarn.lock .
+COPY tsconfig.json .
+COPY .eslintrc .
+
+COPY /packages/rss-handler/package.json ./packages/rss-handler/package.json
+
+RUN yarn install --pure-lockfile
+
+ADD /packages/rss-handler ./packages/rss-handler
+RUN yarn workspace @omnivore/rss-handler build
+
+# After building, fetch the production dependencies
+RUN rm -rf /app/packages/rss-handler/node_modules
+RUN rm -rf /app/node_modules
+RUN yarn install --pure-lockfile --production
+
+EXPOSE 8080
+
+CMD ["yarn", "workspace", "@omnivore/rss-handler", "start"]
+

packages/rss-handler/mocha-config.json

@@ -0,0 +1,5 @@
+{
+    "extension": ["ts"],
+    "spec": "test/**/*.test.ts",
+    "require": "test/babel-register.js"
+  }

packages/rss-handler/package.json

@@ -0,0 +1,30 @@
+{
+  "name": "@omnivore/rss-handler",
+  "version": "1.0.0",
+  "main": "build/src/index.js",
+  "files": [
+    "build/src"
+  ],
+  "license": "Apache-2.0",
+  "scripts": {
+    "test": "yarn mocha -r ts-node/register --config mocha-config.json",
+    "lint": "eslint src --ext ts,js,tsx,jsx",
+    "compile": "tsc",
+    "build": "tsc",
+    "start": "functions-framework --target=rssHandler",
+    "dev": "concurrently \"tsc -w\" \"nodemon --watch ./build/ --exec npm run start\""
+  },
+  "devDependencies": {
+    "chai": "^4.3.6",
+    "eslint-plugin-prettier": "^4.0.0",
+    "mocha": "^10.0.0"
+  },
+  "dependencies": {
+    "@google-cloud/functions-framework": "3.1.2",
+    "@sentry/serverless": "^6.16.1",
+    "axios": "^1.4.0",
+    "dotenv": "^16.0.1",
+    "jsonwebtoken": "^8.5.1",
+    "rss-parser": "^3.13.0"
+  }
+}

packages/rss-handler/src/index.ts

@@ -0,0 +1,209 @@
+import * as Sentry from '@sentry/serverless'
+import axios from 'axios'
+import * as dotenv from 'dotenv' // see https://github.com/motdotla/dotenv#how-do-i-use-dotenv-with-import
+import * as jwt from 'jsonwebtoken'
+import Parser from 'rss-parser'
+import { promisify } from 'util'
+
+interface RssFeedRequest {
+  subscriptionId: string
+  userId: string
+  feedUrl: string
+  lastFetchedAt: Date
+}
+
+function isRssFeedRequest(body: any): body is RssFeedRequest {
+  return (
+    'subscriptionId' in body &&
+    'userId' in body &&
+    'feedUrl' in body &&
+    'lastFetchedAt' in body
+  )
+}
+
+const sendSavePageMutation = async (userId: string, input: unknown) => {
+  const JWT_SECRET = process.env.JWT_SECRET
+  const REST_BACKEND_ENDPOINT = process.env.REST_BACKEND_ENDPOINT
+
+  if (!JWT_SECRET || !REST_BACKEND_ENDPOINT) {
+    throw 'Environment not configured correctly'
+  }
+
+  const data = JSON.stringify({
+    query: `mutation SavePage ($input: SavePageInput!){
+          savePage(input:$input){
+            ... on SaveSuccess{
+              url
+              clientRequestId
+            }
+            ... on SaveError{
+                errorCodes
+            }
+          }
+    }`,
+    variables: {
+      input: Object.assign({}, input, { source: 'puppeteer-parse' }),
+    },
+  })
+
+  const auth = (await signToken({ uid: userId }, JWT_SECRET)) as string
+  try {
+    const response = await axios.post(
+      `${REST_BACKEND_ENDPOINT}/graphql`,
+      data,
+      {
+        headers: {
+          Cookie: `auth=${auth};`,
+          'Content-Type': 'application/json',
+        },
+        timeout: 30000, // 30s
+      }
+    )
+
+    /* eslint-disable @typescript-eslint/no-unsafe-member-access */
+    return !!response.data.data.savePage
+  } catch (error) {
+    if (axios.isAxiosError(error)) {
+      console.error('save page mutation error', error.message)
+    } else {
+      console.error(error)
+    }
+    return false
+  }
+}
+
+const sendUpdateSubscriptionMutation = async (
+  userId: string,
+  subscriptionId: string,
+  lastFetchedAt: Date
+) => {
+  const JWT_SECRET = process.env.JWT_SECRET
+  const REST_BACKEND_ENDPOINT = process.env.REST_BACKEND_ENDPOINT
+
+  if (!JWT_SECRET || !REST_BACKEND_ENDPOINT) {
+    throw 'Environment not configured correctly'
+  }
+
+  const data = JSON.stringify({
+    query: `mutation UpdateSubscription($input: UpdateSubscriptionInput!){
+      updateSubscription(input:$input){
+        ... on UpdateSubscriptionSuccess{
+          subscription{
+            id
+            lastFetchedAt
+          }
+        }
+        ... on UpdateSubscriptionError{
+            errorCodes
+        }
+      }
+    }`,
+    variables: {
+      input: {
+        id: subscriptionId,
+        lastFetchedAt,
+      },
+    },
+  })
+
+  const auth = (await signToken({ uid: userId }, JWT_SECRET)) as string
+  try {
+    const response = await axios.post(
+      `${REST_BACKEND_ENDPOINT}/graphql`,
+      data,
+      {
+        headers: {
+          Cookie: `auth=${auth};`,
+          'Content-Type': 'application/json',
+        },
+        timeout: 30000, // 30s
+      }
+    )
+
+    /* eslint-disable @typescript-eslint/no-unsafe-member-access */
+    return !!response.data.data.savePage
+  } catch (error) {
+    if (axios.isAxiosError(error)) {
+      console.error('update subscription mutation error', error.message)
+    } else {
+      console.error(error)
+    }
+    return false
+  }
+}
+
+dotenv.config()
+Sentry.GCPFunction.init({
+  dsn: process.env.SENTRY_DSN,
+  tracesSampleRate: 0,
+})
+
+const signToken = promisify(jwt.sign)
+const parser = new Parser()
+
+export const rssHandler = Sentry.GCPFunction.wrapHttpFunction(
+  async (req, res) => {
+    if (!process.env.JWT_SECRET) {
+      console.error('Missing JWT_SECRET in environment')
+      return res.status(500).send('INTERNAL_SERVER_ERROR')
+    }
+
+    try {
+      if (!isRssFeedRequest(req.body)) {
+        console.error('Invalid request body', req.body)
+        return res.status(400).send('INVALID_REQUEST_BODY')
+      }
+
+      const { userId, feedUrl, subscriptionId, lastFetchedAt } = req.body
+      // fetch feed
+      const feed = await parser.parseURL(feedUrl)
+      const newFetchedAt = new Date()
+      console.log('Fetched feed', feed.title, newFetchedAt)
+
+      // save each item in the feed
+      for (const item of feed.items) {
+        if (!item.link || !item.title || !item.content || !item.isoDate) {
+          console.log('Invalid feed item', item)
+          continue
+        }
+
+        if (new Date(item.isoDate) <= lastFetchedAt) {
+          console.log('Skipping old feed item', item.title)
+          continue
+        }
+
+        const input = {
+          source: 'rss-feeder',
+          url: item.link,
+          saveRequestId: '',
+          labels: [{ name: 'RSS' }],
+          title: item.title,
+          originalContent: item.content,
+          rssFeedUrl: feedUrl,
+        }
+
+        try {
+          console.log('Saving page', input.title)
+          // save page
+          const result = await sendSavePageMutation(userId, input)
+          console.log('Saved page', result)
+        } catch (error) {
+          console.error('Error while saving page', error)
+        }
+      }
+
+      // update subscription lastFetchedAt
+      const updatedSubscription = await sendUpdateSubscriptionMutation(
+        userId,
+        subscriptionId,
+        newFetchedAt
+      )
+      console.log('Updated subscription', updatedSubscription)
+
+      res.send('ok')
+    } catch (e) {
+      console.error('Error while parsing RSS feed', e)
+      res.status(500).send('INTERNAL_SERVER_ERROR')
+    }
+  }
+)

packages/rss-handler/test/babel-register.js

@@ -0,0 +1,3 @@
+const register = require('@babel/register').default
+
+register({ extensions: ['.ts', '.tsx', '.js', '.jsx'] })

packages/rss-handler/test/stub.test.ts

@@ -0,0 +1,8 @@
+import 'mocha'
+import { expect } from 'chai'
+
+describe('stub test', () => {
+  it('should pass', () => {
+    expect(true).to.be.true
+  })
+})

packages/rss-handler/tsconfig.json

@@ -0,0 +1,8 @@
+{
+  "extends": "./../../tsconfig.json",
+  "compilerOptions": {
+    "outDir": "build",
+    "rootDir": "."
+  },
+  "include": ["src"]
+}

yarn.lock

@@ -13762,7 +13762,7 @@ ent@^2.2.0:
   resolved "https://registry.yarnpkg.com/ent/-/ent-2.2.0.tgz#e964219325a21d05f44466a2f686ed6ce5f5dd1d"
   integrity sha1-6WQhkyWiHQX0RGai9obtbOX13R0=
 
-entities@^2.0.0:
+entities@^2.0.0, entities@^2.0.3:
   version "2.2.0"
   resolved "https://registry.yarnpkg.com/entities/-/entities-2.2.0.tgz#098dc90ebb83d8dffa089d55256b351d34c4da55"
   integrity sha512-p92if5Nz619I0w+akJrLZH0MX0Pb5DX39XOwQTtXSdQQOaYH03S1uIQp4mhOZtAXrxq4ViO67YTiLBo2638o9A==
@@ -24814,6 +24814,14 @@ rollup@2.78.0:
   optionalDependencies:
     fsevents "~2.3.2"
 
+rss-parser@^3.13.0:
+  version "3.13.0"
+  resolved "https://registry.yarnpkg.com/rss-parser/-/rss-parser-3.13.0.tgz#f1f83b0a85166b8310ec531da6fbaa53ff0f50f0"
+  integrity sha512-7jWUBV5yGN3rqMMj7CZufl/291QAhvrrGpDNE4k/02ZchL0npisiYYqULF71jCEKoIiHvK/Q2e6IkDwPziT7+w==
+  dependencies:
+    entities "^2.0.3"
+    xml2js "^0.5.0"
+
 rsvp@^4.8.4, rsvp@^4.8.5:
   version "4.8.5"
   resolved "https://registry.yarnpkg.com/rsvp/-/rsvp-4.8.5.tgz#c8f155311d167f68f21e168df71ec5b083113734"
@@ -28459,6 +28467,14 @@ xml2js@^0.4.23:
     sax ">=0.6.0"
     xmlbuilder "~11.0.0"
 
+xml2js@^0.5.0:
+  version "0.5.0"
+  resolved "https://registry.yarnpkg.com/xml2js/-/xml2js-0.5.0.tgz#d9440631fbb2ed800203fad106f2724f62c493b7"
+  integrity sha512-drPFnkQJik/O+uPKpqSgr22mpuFHqKdbS835iAQrUC73L2F5WkboIRd63ai/2Yg6I1jzifPFKH2NTK+cfglkIA==
+  dependencies:
+    sax ">=0.6.0"
+    xmlbuilder "~11.0.0"
+
 xmlbuilder@~11.0.0:
   version "11.0.1"
   resolved "https://registry.yarnpkg.com/xmlbuilder/-/xmlbuilder-11.0.1.tgz#be9bae1c8a046e76b31127726347d0ad7002beb3"